diff --git a/CHANGELOG.md b/CHANGELOG.md index 021f7697..2298767a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,61 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (docs.yml), `taiki-e/install-action` 2.81.10→2.82.9 (fuzz.yml) — supersedes dependabot #220/#221/#222. +### Changed — BREAKING: WAL v3 is now the only WAL; XactCommit unified to the db-aware record (pre-1.0 format freeze) + +Moon is pre-release: this drops on-disk format compatibility that no shipped +release depended on. There is no migration path — operators upgrading a node +that still has WAL v2 files (`shard-N.wal`) or pre-freeze XactCommit(0x51) +records on disk must let those age out via the AOF-authoritative recovery +path (below), or replay them on a pre-freeze build first and re-persist. + +- **`src/persistence/wal_v3/record.rs`**: deleted the pre-1.0 `XactCommit` + (v1, tag `0x51`) record type, its encoder, and its replay arm. Renamed + `XactCommitV2` -> `XactCommit`, **keeping discriminant `0x53`** (`0x51` is + retired, not reused — a stray `0x51` record in a v3 stream now decodes as + an unrecognized tag rather than being silently reinterpreted). All + producers (the TXN.COMMIT WAL record in `handler_sharded`/`handler_monoio`) + already wrote only the v2 (now unified) record, so this is a pure + simplification of the read side. +- **`src/persistence/wal.rs` deleted** — WAL v2 (`WalWriter`, `replay_wal`, + `wal_path`) is removed in its entirety. WAL v3 (`WalWriterV3`, + `src/persistence/wal_v3/`) is the only WAL, and is now also created in the + old v2 niche (`--appendonly yes` with `--disk-offload disable`), rooted at + `/shard-N/wal-v3/` instead of the old flat + `/shard-N.wal` file — matching the disk-offload layout. + This closes a latent gap: `--appendfsync always` previously only fsynced + WAL v3 (a no-op in non-offload mode, since only v2 existed there); it now + fsyncs the WAL unconditionally. +- **`replay_wal_auto`** rejects a v2-format WAL file (`RRDWAL` magic + + version byte `2`) with a loud `WalError::UnsupportedVersion` instead of + delegating to the deleted v2 replay path. +- **Recovery**: the `shard-N.wal` (v2) fallback rung is removed from both + `persistence::recovery::recover_shard_v3_pitr` (disk-offload path) and + `Shard::restore_from_persistence_v2` (legacy path) — `appendonly.aof` + remains the recovery authority and is unaffected. `shared_databases:: + replay_graph_wal` (graph-feature boot-time WAL scan) is ported from the v2 + flat file to the v3 segment directory (`shard-N/wal-v3/*.wal`), matching + `replay_workspace_wal`/`replay_temporal_wal`. Additionally, both legacy-dir + recovery paths gain a **last-resort WAL v3 fallback**: when NO + `appendonly.aof` exists, the legacy-mode `shard-N/wal-v3/` directory is + replayed (with a loud partial-coverage warning — WAL v3's KV coverage is + intentionally partial post-#211, so it never shadows a present AOF), and a + leftover legacy `shard-N.wal` (v2) file on disk now triggers a + `tracing::error!` naming the file instead of being silently ignored. +- **`wal_append_and_fanout` / `wal_fanout_has_work`** (`src/shard/ + spsc_handler.rs`) drop the v2 `wal_writer` parameter and the "v3 + supersedes v2" branch — a single `Option` parameter, renamed + `wal_writer`. `finalize_snapshot_success` no longer takes a WAL writer: + v2's per-snapshot `truncate_after_snapshot(epoch)` had no v3 analogue — v3 + retention is LSN-driven (`WalWriterV3::recycle_aggressive` / + `recycle_segments_before`, run from autovacuum Pass C and the checkpoint + protocol) and already ran independently of legacy snapshot epochs, so + nothing was lost by not porting it over. +- Deferred (unrelated to this change, found in passing): `shared_databases:: + replay_temporal_wal` scans `shard-N/` directly for `*.wal` files instead of + `shard-N/wal-v3/` like its siblings — looks like a pre-existing directory + mismatch, left as-is pending its own investigation. + ### Fixed — TopLevel-monoio AOF writer: EverySec fsync deferred indefinitely when idle (PR #TBD) - **`src/persistence/aof/writer_task.rs`**: the TopLevel monoio AOF writer diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index cdbcc326..9b6d3e55 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -17,5 +17,4 @@ pub mod redis_rdb; pub mod replay; pub mod snapshot; pub mod vec_undo; -pub mod wal; pub mod wal_v3; diff --git a/src/persistence/recovery.rs b/src/persistence/recovery.rs index 8b0c1a59..c7cba2e5 100644 --- a/src/persistence/recovery.rs +++ b/src/persistence/recovery.rs @@ -562,47 +562,88 @@ pub fn recover_shard_v3_pitr( } } - // ── Phase 4b: V2 WAL FALLBACK ────────────────────────────────────── - // When v3 replay produced 0 commands and a v2 persistence directory is - // available, fall back to replaying the v2 AOF file. This handles the - // common case where --disk-offload enable was used with --appendonly yes - // but write commands logged to the v2 AOF (standard appendonly path). + // Loud failure (not silent skip) for a leftover pre-freeze WAL v2 file + // in the legacy persistence dir. WAL v2 support was removed in this + // build; the file is never consulted by any recovery path here, so + // surface it unconditionally (not only when the branch below happens to + // run) so an operator who upgraded over a v2-only deployment finds out + // immediately rather than discovering a silent recovery gap later. + if let Some(v2_dir) = v2_persistence_dir { + let legacy_v2_wal = v2_dir.join(format!("shard-{}.wal", shard_id)); + if legacy_v2_wal.exists() { + tracing::error!( + "Shard {}: found legacy WAL v2 file {:?} — WAL v2 support was \ + removed (pre-1.0 WAL-v3-only format freeze) and this build \ + does NOT replay it. If its contents are not already reflected \ + in appendonly.aof, replay it on a pre-freeze Moon build first \ + (which re-persists through the AOF) before removing this file.", + shard_id, + legacy_v2_wal + ); + } + } + + // ── Phase 4b: legacy-dir AOF / WAL v3 FALLBACK ───────────────────── + // When the disk-offload WAL v3 replay (Phase 4, above) produced 0 + // commands and a legacy persistence directory is available, fall back to + // it. `appendonly.aof` is the AUTHORITY there (post-#211, WAL v3's KV + // coverage is intentionally partial: `--wal-kv-log` default-off, and + // connection-local writes bypass it even when on — a non-empty WAL v3 + // must never shadow the AOF). Only when NO AOF exists at all does the + // legacy-mode WAL v3 directory (`v2_dir/shard-N/wal-v3/`, distinct from + // `shard_dir`'s Phase-4 WAL despite the shared name) serve as a + // last-resort partial recovery source. The WAL v2 rung (per-shard + // `shard-N.wal`) was removed in the pre-1.0 WAL-v3-only format freeze + // and is never replayed by this build. if result.commands_replayed == 0 { if let Some(v2_dir) = v2_persistence_dir { - // Try all v2 persistence sources in order: - // 1. Per-shard binary WAL (shard-N.wal) - // 2. Global RESP-format AOF (appendonly.aof) - let v2_sources: &[(&std::path::Path, bool)] = &[ - (&crate::persistence::wal::wal_path(v2_dir, shard_id), false), - (&v2_dir.join("appendonly.aof"), true), - ]; - for &(ref path, is_aof) in v2_sources { - if !path.exists() { - continue; - } + let aof_path = v2_dir.join("appendonly.aof"); + if aof_path.exists() { info!( - "Shard {}: v3 WAL empty, falling back to v2 replay from {:?}", - shard_id, path + "Shard {}: WAL empty, falling back to AOF replay from {:?}", + shard_id, aof_path ); - let replay_result = if is_aof { - crate::persistence::aof::replay_aof(databases, path, engine) - } else { - crate::persistence::wal::replay_wal(databases, path, engine) - }; - match replay_result { - Ok(n) if n > 0 => { + match crate::persistence::aof::replay_aof(databases, &aof_path, engine) { + Ok(n) => { result.commands_replayed = n; - info!("Shard {}: v2 fallback replayed {} commands", shard_id, n); - break; + info!("Shard {}: AOF fallback replayed {} commands", shard_id, n); } - Ok(_) => { - info!( - "Shard {}: v2 source {:?} had 0 commands, trying next", - shard_id, path + Err(e) => { + tracing::error!( + "Shard {}: AOF fallback {:?} failed: {}", + shard_id, + aof_path, + e + ); + } + } + } else { + let legacy_wal_v3_dir = v2_dir.join(format!("shard-{}", shard_id)).join("wal-v3"); + match crate::persistence::wal_v3::replay::replay_wal_v3_dir_commands( + &legacy_wal_v3_dir, + databases, + engine, + ) { + Ok(0) => {} + Ok(n) => { + result.commands_replayed = n; + tracing::warn!( + "Shard {}: no appendonly.aof found — replayed {} records \ + from legacy-mode WAL v3 at {:?} as a LAST-RESORT \ + fallback. WAL v3 KV coverage is partial (gated by \ + --wal-kv-log; connection-local writes bypass it), so \ + this recovery may be incomplete.", + shard_id, + n, + legacy_wal_v3_dir ); } Err(e) => { - tracing::error!("Shard {}: v2 fallback {:?} failed: {}", shard_id, path, e); + tracing::error!( + "Shard {}: legacy-mode WAL v3 fallback replay failed: {}", + shard_id, + e + ); } } } @@ -1134,4 +1175,161 @@ mod tests { assert_eq!(result.cold_segments[0].0, 50); assert_eq!(result.cold_segments[0].1, seg_dir); } + + // ── Legacy-dir AOF authority + WAL v3 last-resort fallback (review + // finding P0#1, corrected) ────────────────────────────────────────────── + // + // The AOF is the recovery authority for the legacy dir: post-#211, + // WAL v3's KV coverage is intentionally partial (`--wal-kv-log` + // default-off; connection-local writes bypass it even when on), so a + // non-empty legacy-mode WAL v3 directory (`v2_dir/shard-N/wal-v3/`) must + // never shadow `appendonly.aof`. WAL v3 is only a last-resort source + // when NO AOF exists. A raw SIGKILL integration test cannot exercise + // this deterministically (see the NOTE in + // `tests/crash_matrix_per_shard_aof.rs`), so these test the exact + // function directly instead -- the same style every other test in this + // module already uses (construct WAL v3 segments by hand via + // `write_wal_v3_record`, assert on `commands_replayed`). + + #[test] + fn test_legacy_aof_is_authority_over_wal_v3() { + let tmp = tempfile::tempdir().unwrap(); + // Disk-offload shard_dir: empty (no wal-v3 data) so Phase 4 replays 0 + // commands and Phase 4b's fallback engages. + let shard_dir = tmp.path().join("shard-0"); + std::fs::create_dir_all(&shard_dir).unwrap(); + + // Legacy (non-disk-offload) persistence dir: has BOTH a populated + // `shard-0/wal-v3/` (5 records) and a shorter `appendonly.aof` + // (2 records). Post-#211 the WAL routinely holds MORE records than + // the AOF while missing KV writes entirely — the AOF must win. + let v2_dir = tmp.path().join("legacy"); + let legacy_wal_dir = v2_dir.join("shard-0").join("wal-v3"); + std::fs::create_dir_all(&legacy_wal_dir).unwrap(); + let mut wal_data = make_v3_header(0); + for i in 1..=5u64 { + write_wal_v3_record( + &mut wal_data, + i, + WalRecordType::Command, + b"*1\r\n$4\r\nPING\r\n", + ); + } + std::fs::write(legacy_wal_dir.join("000000000001.wal"), &wal_data).unwrap(); + + std::fs::create_dir_all(&v2_dir).unwrap(); + let aof_payload = b"*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n"; // 2 records + std::fs::write(v2_dir.join("appendonly.aof"), aof_payload).unwrap(); + + let mut databases = vec![Database::new()]; + let engine = crate::persistence::replay::DispatchReplayEngine::new(); + let result = + recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir)) + .unwrap(); + + assert_eq!( + result.commands_replayed, 2, + "appendonly.aof (2 records, the authority) must be replayed and \ + the KV-incomplete legacy WAL v3 (5 records) ignored -- got {} \ + (the WAL shadowed the AOF if this is 5)", + result.commands_replayed + ); + } + + #[test] + fn test_legacy_wal_v3_last_resort_when_no_aof() { + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path().join("shard-0"); + std::fs::create_dir_all(&shard_dir).unwrap(); + + // Legacy dir has ONLY a populated WAL v3 — no appendonly.aof at all + // (lost/rebuilt). Partial recovery beats none. + let v2_dir = tmp.path().join("legacy"); + let legacy_wal_dir = v2_dir.join("shard-0").join("wal-v3"); + std::fs::create_dir_all(&legacy_wal_dir).unwrap(); + let mut wal_data = make_v3_header(0); + for i in 1..=5u64 { + write_wal_v3_record( + &mut wal_data, + i, + WalRecordType::Command, + b"*1\r\n$4\r\nPING\r\n", + ); + } + std::fs::write(legacy_wal_dir.join("000000000001.wal"), &wal_data).unwrap(); + + let mut databases = vec![Database::new()]; + let engine = crate::persistence::replay::DispatchReplayEngine::new(); + let result = + recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir)) + .unwrap(); + + assert_eq!( + result.commands_replayed, 5, + "with no appendonly.aof, the legacy WAL v3 last-resort fallback \ + must replay what it can" + ); + } + + #[test] + fn test_legacy_wal_v3_absent_falls_back_to_aof() { + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path().join("shard-0"); + std::fs::create_dir_all(&shard_dir).unwrap(); + + // Legacy dir has ONLY an AOF -- no `shard-0/wal-v3/` at all (the + // common case: disk-offload was never used and `--wal-kv-log` never + // populated WAL v3, so the AOF is genuinely the only source). + let v2_dir = tmp.path().join("legacy"); + std::fs::create_dir_all(&v2_dir).unwrap(); + let aof_payload = b"*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n"; + std::fs::write(v2_dir.join("appendonly.aof"), aof_payload).unwrap(); + + let mut databases = vec![Database::new()]; + let engine = crate::persistence::replay::DispatchReplayEngine::new(); + let result = + recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir)) + .unwrap(); + + assert_eq!( + result.commands_replayed, 3, + "with no legacy WAL v3 data at all, the AOF fallback must still \ + work exactly as before this fix" + ); + } + + #[test] + fn test_stray_legacy_v2_wal_file_does_not_break_recovery() { + // A leftover pre-freeze WAL v2 file (`shard-0.wal`, the old flat + // `RRDWAL` version=2 format) must not be replayed (WAL v2 support was + // removed) and must not crash or otherwise break recovery from the + // AOF -- it should just be loudly logged (see the `tracing::error!` + // in the Phase 4b block above) and ignored. + let tmp = tempfile::tempdir().unwrap(); + let shard_dir = tmp.path().join("shard-0"); + std::fs::create_dir_all(&shard_dir).unwrap(); + + let v2_dir = tmp.path().join("legacy"); + std::fs::create_dir_all(&v2_dir).unwrap(); + // Stray pre-freeze v2 WAL file: RRDWAL magic + version=2 header. + let mut legacy_v2_wal = vec![0u8; 32]; + legacy_v2_wal[0..6].copy_from_slice(b"RRDWAL"); + legacy_v2_wal[6] = 2; + std::fs::write(v2_dir.join("shard-0.wal"), &legacy_v2_wal).unwrap(); + // A real AOF is still present and must still be used. + let aof_payload = b"*1\r\n$4\r\nPING\r\n"; + std::fs::write(v2_dir.join("appendonly.aof"), aof_payload).unwrap(); + + let mut databases = vec![Database::new()]; + let engine = crate::persistence::replay::DispatchReplayEngine::new(); + let result = + recover_shard_v3_with_fallback(&mut databases, 0, &shard_dir, &engine, Some(&v2_dir)) + .unwrap(); + + assert_eq!( + result.commands_replayed, 1, + "a stray legacy v2 WAL file must not prevent the AOF fallback \ + from working" + ); + } } diff --git a/src/persistence/wal.rs b/src/persistence/wal.rs deleted file mode 100644 index bf6f8523..00000000 --- a/src/persistence/wal.rs +++ /dev/null @@ -1,958 +0,0 @@ -//! Per-shard Write-Ahead Log (WAL) writer -- v2 format. -//! -//! Each shard maintains its own WAL file containing RESP-encoded write commands -//! wrapped in CRC32-checksummed block frames with a 32-byte versioned header. -//! -//! **WAL v2 on-disk format:** -//! ```text -//! [Header: 32 bytes] -//! magic: "RRDWAL" (6B) -//! version: u8 = 2 (1B) -//! shard_id: u16 LE (2B) -//! epoch: u64 LE (8B) -//! reserved: [0u8; 15] (15B) -//! -//! [Block frame] (repeated per flush) -//! block_len: u32 LE -- covers cmd_count + db_idx + payload + crc32 -//! cmd_count: u16 LE -//! db_idx: u8 -//! payload: [u8; payload_len] -- raw RESP bytes -//! crc32: u32 LE -- CRC32 over cmd_count(2B) + db_idx(1B) + payload -//! ``` -//! -//! Writes are buffered in memory and flushed/fsynced on a configurable interval -//! (default 1ms for batched fsync). On Linux, uses direct file I/O; on macOS, -//! uses std::fs as fallback. -//! -//! Replaces the global AOF writer from Phase 11 with per-shard locality. -//! -//! ## Unwrap Classification -//! -//! | Context | Classification | Rationale | -//! |---------|---------------|-----------| -//! | `WalWriter` methods (`new`, `flush_*`, `do_write`, etc.) | **must-panic** (`std::io::Result`) | Flush failure = silent data loss | -//! | `replay_wal` / `replay_wal_v2` | **should-recover** (`Result<_, MoonError>`) | Replay is startup-only; failure logs + continues | -//! | `#[cfg(test)]` code (81 unwraps) | **test-only** | Panics are appropriate in tests | - -use std::io::Write; -use std::path::{Path, PathBuf}; -use std::time::Instant; - -use bytes::BytesMut; -use tracing::{info, warn}; - -use crate::error::{MoonError, WalError}; - -use crate::runtime::{FileIoImpl, traits::FileIo}; -use crate::storage::db::Database; - -/// WAL v2 format constants -const WAL_MAGIC: &[u8; 6] = b"RRDWAL"; -const WAL_VERSION: u8 = 2; -const WAL_HEADER_SIZE: usize = 32; -/// Block overhead: block_len(4) + cmd_count(2) + db_idx(1) + crc32(4) = 11 bytes -#[allow(dead_code)] -const BLOCK_OVERHEAD: usize = 11; - -/// Per-shard WAL writer with batched fsync. -/// -/// Accumulates RESP-encoded command bytes between fsync intervals. -/// Flush is called explicitly by the shard event loop on its 1ms tick. -/// Produces CRC32-checksummed block frames for corruption detection. -pub struct WalWriter { - shard_id: usize, - /// Buffered RESP bytes awaiting flush. - buf: Vec, - /// Staging buffer for assembling the full block frame - /// (header + payload + CRC) so each flush is a single write syscall. - /// Cleared after every flush; capacity is retained. - frame_buf: Vec, - /// File handle for the WAL. - file: Option, - /// Path to the WAL file. - file_path: PathBuf, - /// Current write offset in the file. - write_offset: u64, - /// Timestamp of last fsync. - last_fsync: Instant, - /// Current snapshot epoch (for WAL truncation after snapshot). - epoch: u64, - /// Total bytes written since last truncation. - bytes_written: u64, - /// Number of commands appended since last flush (for block framing). - cmd_count: u16, - /// Whether the v2 header has been written to the current file. - header_written: bool, -} - -impl WalWriter { - /// Create a new WAL writer for the given shard. - /// - /// WAL file path: `{dir}/shard-{shard_id}.wal` - /// Opens file in append+create mode. Pre-allocates 8KB buffer. - /// Writes v2 header on fresh files. - pub fn new(shard_id: usize, dir: &Path) -> std::io::Result { - let file_path = wal_path(dir, shard_id); - let file = FileIoImpl::open_append(&file_path)?; - - let write_offset = file.metadata()?.len(); - let header_written = write_offset > 0; - - let mut writer = Self { - shard_id, - buf: Vec::with_capacity(8192), - frame_buf: Vec::with_capacity(8192), - file: Some(file), - file_path, - write_offset, - last_fsync: Instant::now(), - epoch: 0, - bytes_written: 0, - cmd_count: 0, - header_written, - }; - - // Write v2 header on fresh files - if write_offset == 0 { - writer.write_header()?; - } - - Ok(writer) - } - - /// Write the 32-byte v2 header directly to the file. - /// - /// Layout: RRDWAL(6B) + version(1B) + shard_id(2B LE) + epoch(8B LE) + reserved(15B) - fn write_header(&mut self) -> std::io::Result<()> { - if let Some(ref mut file) = self.file { - Self::write_header_to(file, self.shard_id, self.epoch)?; - self.write_offset += WAL_HEADER_SIZE as u64; - self.bytes_written += WAL_HEADER_SIZE as u64; - self.header_written = true; - Ok(()) - } else { - Err(std::io::Error::other("WAL file handle is closed")) - } - } - - /// Serialize and write the v2 header to a file handle. - /// Shared by `write_header()` and the inline fallback in `do_write()`. - fn write_header_to( - file: &mut std::fs::File, - shard_id: usize, - epoch: u64, - ) -> std::io::Result<()> { - let mut header = [0u8; WAL_HEADER_SIZE]; - header[0..6].copy_from_slice(WAL_MAGIC); - header[6] = WAL_VERSION; - header[7..9].copy_from_slice(&(shard_id as u16).to_le_bytes()); - header[9..17].copy_from_slice(&epoch.to_le_bytes()); - // bytes 17..32 remain zero (reserved) - file.write_all(&header) - } - - /// Append RESP-encoded command bytes to the in-memory buffer. - /// - /// No I/O occurs here -- this is the hot path called on every write command. - /// Only adds a saturating_add for cmd_count tracking. - #[inline] - pub fn append(&mut self, data: &[u8]) { - self.buf.extend_from_slice(data); - self.cmd_count = self.cmd_count.saturating_add(1); - } - - /// Append a pre-serialized vector WAL record frame to the WAL buffer. - /// - /// The frame bytes include the VECTOR_RECORD_TAG, length, payload, and CRC. - /// This is NOT wrapped in a RESP block frame -- it's a standalone frame type - /// that the WAL reader identifies by its first byte (0x56 vs block_len). - /// - /// Called by vector command handlers after mutation. - /// Does NOT increment cmd_count -- vector records are not RESP commands. - #[inline] - pub fn append_vector_record(&mut self, frame_bytes: &[u8]) { - self.buf.extend_from_slice(frame_bytes); - } - - /// Flush buffered data to OS page cache if the buffer is non-empty. - /// - /// Called on the shard's 1ms tick. Only does write_all() (fast, goes to - /// OS page cache), NOT fsync. Use sync_to_disk() for durability. - pub fn flush_if_needed(&mut self) -> std::io::Result<()> { - if self.buf.is_empty() { - return Ok(()); - } - self.do_write() - } - - /// Force flush with fsync. Used on shutdown and before snapshot. - pub fn flush_sync(&mut self) -> std::io::Result<()> { - if self.buf.is_empty() { - return Ok(()); - } - self.do_write()?; - self.do_sync() - } - - /// Sync file data to disk (fsync). Called on a less-frequent interval - /// (e.g. every second for "everysec" mode) to avoid blocking the shard - /// on every 1ms tick. - pub fn sync_to_disk(&mut self) -> std::io::Result<()> { - if let Some(ref mut file) = self.file { - file.sync_data()?; - self.last_fsync = Instant::now(); - Ok(()) - } else { - Ok(()) // No file open, nothing to sync - } - } - - /// Internal write: produce a CRC32-framed block from the buffer, write to file. - /// - /// Block layout: [block_len:4 LE][cmd_count:2 LE][db_idx:1][payload:var][crc32:4 LE] - /// CRC32 covers: cmd_count(2B) + db_idx(1B) + payload bytes. - fn do_write(&mut self) -> std::io::Result<()> { - if let Some(ref mut file) = self.file { - // Ensure header is written before any block data - if !self.header_written { - Self::write_header_to(file, self.shard_id, self.epoch)?; - self.write_offset += WAL_HEADER_SIZE as u64; - self.bytes_written += WAL_HEADER_SIZE as u64; - self.header_written = true; - } - - let cmd_count_bytes = self.cmd_count.to_le_bytes(); - // db_idx is always 0: in per-shard architecture, each shard owns its - // own database slice and SELECT commands are replayed from the RESP - // payload during recovery, so the block-level db_idx is not used for - // routing. The field is retained in the format for forward compatibility. - let db_idx: u8 = 0; - - // Compute CRC32 over cmd_count(2B) + db_idx(1B) + payload - let mut hasher = crc32fast::Hasher::new(); - hasher.update(&cmd_count_bytes); - hasher.update(&[db_idx]); - hasher.update(&self.buf); - let crc = hasher.finalize(); - - // block_len = cmd_count(2) + db_idx(1) + payload.len() + crc32(4) - let block_len = (2 + 1 + self.buf.len() + 4) as u32; - - // Assemble the full block frame (header + payload + CRC) in the - // reusable staging buffer so the flush is ONE write syscall - // instead of three. On-disk layout is unchanged: - // [block_len:4 LE][cmd_count:2 LE][db_idx:1][payload:var][crc32:4 LE] - self.frame_buf.clear(); - self.frame_buf.reserve(7 + self.buf.len() + 4); - self.frame_buf.extend_from_slice(&block_len.to_le_bytes()); - self.frame_buf.extend_from_slice(&cmd_count_bytes); - self.frame_buf.push(db_idx); - self.frame_buf.extend_from_slice(&self.buf); - self.frame_buf.extend_from_slice(&crc.to_le_bytes()); - - file.write_all(&self.frame_buf)?; - - let total_written = 4 + block_len as u64; // block_len prefix + block_len contents - self.write_offset += total_written; - self.bytes_written += total_written; - self.cmd_count = 0; - self.buf.clear(); // clear but keep allocation - Ok(()) - } else { - Err(std::io::Error::other("WAL file handle is closed")) - } - } - - /// Internal sync: fsync only. - fn do_sync(&mut self) -> std::io::Result<()> { - if let Some(ref mut file) = self.file { - file.sync_data()?; - self.last_fsync = Instant::now(); - Ok(()) - } else { - Err(std::io::Error::other("WAL file handle is closed")) - } - } - - /// Truncate WAL after snapshot completion. - /// - /// Flushes pending data, renames current WAL to `.wal.old` (backup), - /// then opens a fresh WAL file with v2 header. Resets counters and updates epoch. - pub fn truncate_after_snapshot(&mut self, new_epoch: u64) -> std::io::Result<()> { - // Flush any pending data first - if !self.buf.is_empty() { - self.flush_sync()?; - } - - // Close current file - self.file.take(); - - // Rename current WAL to .wal.old (overwrites previous backup) - let old_path = self.file_path.with_extension("wal.old"); - if self.file_path.exists() { - std::fs::rename(&self.file_path, &old_path)?; - if let Some(parent) = self.file_path.parent() { - crate::persistence::fsync::fsync_directory(parent)?; - } - } - - // Open a fresh WAL file - let file = FileIoImpl::open_append(&self.file_path)?; - self.file = Some(file); - - // Reset counters - self.write_offset = 0; - self.bytes_written = 0; - self.epoch = new_epoch; - self.cmd_count = 0; - self.header_written = false; - - // Write v2 header on the fresh file - self.write_header()?; - - info!( - "WAL truncated for shard {} at epoch {}", - self.shard_id, new_epoch - ); - Ok(()) - } - - /// Graceful shutdown: flush pending data, fsync, close file handle. - pub fn shutdown(&mut self) -> std::io::Result<()> { - if !self.buf.is_empty() { - self.flush_sync()?; - } - // Drop file handle explicitly - self.file.take(); - info!("WAL writer for shard {} shut down", self.shard_id); - Ok(()) - } - - /// Returns the path to the WAL file. - pub fn path(&self) -> &Path { - &self.file_path - } - - /// Returns total bytes written since last truncation. - pub fn bytes_written(&self) -> u64 { - self.bytes_written - } -} - -/// Replay a per-shard WAL file into the given databases. -/// -/// Auto-detects v1 (raw RESP) vs v2 (RRDWAL header + CRC32 block framing) format -/// by checking whether the first 6 bytes match the WAL_MAGIC signature. -/// -/// Returns the number of commands successfully replayed. -pub fn replay_wal( - databases: &mut [Database], - path: &Path, - engine: &dyn crate::persistence::replay::CommandReplayEngine, -) -> Result { - let data = std::fs::read(path)?; - if data.is_empty() { - return Ok(0); - } - // Auto-detect: check if first 6 bytes match WAL_MAGIC - if data.len() >= WAL_HEADER_SIZE && &data[..6] == WAL_MAGIC { - replay_wal_v2(databases, &data, engine) - } else { - // V1 fallback: delegate to AOF replay (raw RESP) - crate::persistence::aof::replay_aof(databases, path, engine) - } -} - -/// Replay a WAL v2 file from an in-memory byte slice. -/// -/// Parses the 32-byte header, then iterates over CRC32-checksummed block frames. -/// Stops on first corrupted or truncated block, returning commands replayed so far. -fn replay_wal_v2( - databases: &mut [Database], - data: &[u8], - engine: &dyn crate::persistence::replay::CommandReplayEngine, -) -> Result { - // Parse and validate header - if data.len() < WAL_HEADER_SIZE { - return Err(WalError::Corrupted { - offset: 0, - detail: "WAL v2 file too short for header".into(), - } - .into()); - } - let version = data[6]; - if version != WAL_VERSION { - return Err(WalError::UnsupportedVersion { - version: version.into(), - } - .into()); - } - let shard_id = u16::from_le_bytes([data[7], data[8]]); - let epoch = u64::from_le_bytes(data[9..17].try_into().map_err(|_| { - MoonError::from(WalError::Corrupted { - offset: 9, - detail: "invalid epoch bytes at header offset 9..17".into(), - }) - })?); - info!("Replaying WAL v2: shard={}, epoch={}", shard_id, epoch); - - let mut offset = WAL_HEADER_SIZE; - let mut total_commands: usize = 0; - let config = crate::protocol::ParseConfig::default(); - let mut selected_db: usize = 0; - - while offset < data.len() { - // Need at least 4 bytes for block_len - if offset + 4 > data.len() { - warn!("WAL v2: truncated block_len at offset {}, stopping", offset); - break; - } - let block_len = u32::from_le_bytes(data[offset..offset + 4].try_into().map_err(|_| { - MoonError::from(WalError::Corrupted { - offset: offset as u64, - detail: format!("invalid block_len bytes at offset {}", offset), - }) - })?) as usize; - offset += 4; - - // Minimum block content: cmd_count(2) + db_idx(1) + crc32(4) = 7 - if block_len < 7 { - warn!( - "WAL v2: invalid block_len {} at offset {}, stopping", - block_len, - offset - 4 - ); - break; - } - if offset + block_len > data.len() { - warn!( - "WAL v2: block extends past EOF at offset {}, stopping", - offset - 4 - ); - break; - } - - let block_data = &data[offset..offset + block_len]; - - // Extract fields - let payload_len = block_len - 7; // minus cmd_count(2) + db_idx(1) + crc32(4) - let payload = &block_data[3..3 + payload_len]; - let stored_crc = u32::from_le_bytes( - block_data[block_len - 4..block_len] - .try_into() - .map_err(|_| { - MoonError::from(WalError::Corrupted { - offset: (offset - 4) as u64, - detail: format!("invalid CRC bytes at block offset {}", offset - 4), - }) - })?, - ); - - // Verify CRC: covers cmd_count(2B) + db_idx(1B) + payload - let mut hasher = crc32fast::Hasher::new(); - hasher.update(&block_data[0..3]); // cmd_count + db_idx - hasher.update(payload); - let computed_crc = hasher.finalize(); - - if computed_crc != stored_crc { - warn!( - "WAL v2: CRC mismatch at offset {} (stored={:#010x}, computed={:#010x}), \ - stopping replay after {} commands", - offset - 4, - stored_crc, - computed_crc, - total_commands - ); - break; // Stop on first corrupted block -- ordering matters - } - - // Parse RESP commands from payload - let mut buf = BytesMut::from(payload); - - while !buf.is_empty() { - match crate::protocol::parse::parse(&mut buf, &config) { - Ok(Some(frame)) => { - let (cmd, cmd_args) = match &frame { - crate::protocol::Frame::Array(arr) if !arr.is_empty() => { - let name = match &arr[0] { - crate::protocol::Frame::BulkString(s) => s.as_ref(), - crate::protocol::Frame::SimpleString(s) => s.as_ref(), - _ => { - total_commands += 1; - continue; - } - }; - (name as &[u8], &arr[1..]) - } - _ => { - total_commands += 1; - continue; - } - }; - engine.replay_command(databases, cmd, cmd_args, &mut selected_db); - total_commands += 1; - } - Ok(None) => { - if !buf.is_empty() { - warn!("WAL v2: truncated RESP in block at offset {}", offset - 4); - } - break; - } - Err(e) => { - warn!("WAL v2: RESP parse error in block: {}", e); - break; - } - } - } - - offset += block_len; - } - - Ok(total_commands) -} - -/// Construct the WAL file path for a given shard. -pub fn wal_path(dir: &Path, shard_id: usize) -> PathBuf { - dir.join(format!("shard-{}.wal", shard_id)) -} - -#[cfg(test)] -mod tests { - use super::*; - use bytes::{Bytes, BytesMut}; - use tempfile::tempdir; - - use crate::persistence::aof::serialize_command; - use crate::persistence::replay::DispatchReplayEngine; - use crate::protocol::Frame; - use crate::protocol::serialize; - - fn make_command(parts: &[&[u8]]) -> Frame { - Frame::Array( - parts - .iter() - .map(|p| Frame::BulkString(Bytes::copy_from_slice(p))) - .collect(), - ) - } - - fn serialize_to_bytes(frame: &Frame) -> Vec { - let mut buf = BytesMut::new(); - serialize::serialize(frame, &mut buf); - buf.to_vec() - } - - #[test] - fn test_wal_append_and_flush() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - let cmd = make_command(&[b"SET", b"k", b"v"]); - let data = serialize_to_bytes(&cmd); - - writer.append(&data); - writer.flush_if_needed().unwrap(); - - // Read file back -- v2 format: header(32) + block frame - let contents = std::fs::read(writer.path()).unwrap(); - // File starts with RRDWAL header - assert_eq!(&contents[0..6], b"RRDWAL"); - assert_eq!(contents[6], WAL_VERSION); - // After header, the block contains the RESP payload - let block_start = WAL_HEADER_SIZE; - let block_len = - u32::from_le_bytes(contents[block_start..block_start + 4].try_into().unwrap()); - // Extract payload from block: skip block_len(4) + cmd_count(2) + db_idx(1) = 7 bytes - let payload_start = block_start + 7; - let payload_len = block_len as usize - 2 - 1 - 4; // subtract cmd_count + db_idx + crc - let payload = &contents[payload_start..payload_start + payload_len]; - assert_eq!(payload, &data[..]); - } - - #[test] - fn test_wal_replay_round_trip() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Append multiple SET commands via serialize_command - let cmd1 = make_command(&[b"SET", b"key1", b"val1"]); - let cmd2 = make_command(&[b"SET", b"key2", b"val2"]); - let cmd3 = make_command(&[b"SET", b"key3", b"val3"]); - - writer.append(&serialize_command(&cmd1)); - writer.append(&serialize_command(&cmd2)); - writer.append(&serialize_command(&cmd3)); - writer.flush_sync().unwrap(); - - // Replay into databases - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, writer.path(), &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 3); - - assert_eq!( - dbs[0].get(b"key1").unwrap().value.as_bytes().unwrap(), - b"val1" - ); - assert_eq!( - dbs[0].get(b"key2").unwrap().value.as_bytes().unwrap(), - b"val2" - ); - assert_eq!( - dbs[0].get(b"key3").unwrap().value.as_bytes().unwrap(), - b"val3" - ); - } - - #[test] - fn test_wal_truncate_after_snapshot() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Write some data - let cmd = make_command(&[b"SET", b"k", b"v"]); - let data = serialize_to_bytes(&cmd); - writer.append(&data); - writer.flush_sync().unwrap(); - - // Verify data was written (header + block) - let contents_before = std::fs::read(writer.path()).unwrap(); - assert!(!contents_before.is_empty()); - assert_eq!(&contents_before[0..6], b"RRDWAL"); - - // Truncate - writer.truncate_after_snapshot(1).unwrap(); - - // New WAL file should have just the v2 header (32 bytes) - let contents_after = std::fs::read(writer.path()).unwrap(); - assert_eq!(contents_after.len(), WAL_HEADER_SIZE); - assert_eq!(&contents_after[0..6], b"RRDWAL"); - // Epoch in the new header should be 1 - let epoch = u64::from_le_bytes(contents_after[9..17].try_into().unwrap()); - assert_eq!(epoch, 1); - - // Old data should be in .wal.old - let old_path = writer.path().with_extension("wal.old"); - let old_contents = std::fs::read(&old_path).unwrap(); - assert_eq!(&old_contents[0..6], b"RRDWAL"); - - // bytes_written includes header bytes - assert_eq!(writer.bytes_written(), WAL_HEADER_SIZE as u64); - } - - #[test] - fn test_wal_flush_empty_is_noop() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Flush on empty buffer should return Ok without any block I/O - writer.flush_if_needed().unwrap(); - writer.flush_sync().unwrap(); - - // File should contain only the header (written in new()) - let contents = std::fs::read(writer.path()).unwrap(); - assert_eq!(contents.len(), WAL_HEADER_SIZE); - assert_eq!(&contents[0..6], b"RRDWAL"); - } - - #[test] - fn test_wal_multiple_appends_batched() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Append 100 small commands - let mut expected_payload = Vec::new(); - for i in 0..100 { - let key = format!("k{}", i); - let val = format!("v{}", i); - let cmd = make_command(&[b"SET", key.as_bytes(), val.as_bytes()]); - let data = serialize_to_bytes(&cmd); - expected_payload.extend_from_slice(&data); - writer.append(&data); - } - - // Single flush - writer.flush_sync().unwrap(); - - // File starts with header, then a single block containing all 100 commands - let contents = std::fs::read(writer.path()).unwrap(); - assert_eq!(&contents[0..6], b"RRDWAL"); - - // Parse the block - let block_start = WAL_HEADER_SIZE; - let block_len = - u32::from_le_bytes(contents[block_start..block_start + 4].try_into().unwrap()); - let cmd_count = u16::from_le_bytes( - contents[block_start + 4..block_start + 6] - .try_into() - .unwrap(), - ); - assert_eq!(cmd_count, 100); - - // Extract payload - let payload_start = block_start + 7; - let payload_len = block_len as usize - 2 - 1 - 4; - let payload = &contents[payload_start..payload_start + payload_len]; - assert_eq!(payload, &expected_payload[..]); - - assert!(writer.bytes_written() > 0); - } - - #[test] - fn test_wal_path_construction() { - let dir = Path::new("/tmp/wal"); - assert_eq!(wal_path(dir, 0), PathBuf::from("/tmp/wal/shard-0.wal")); - assert_eq!(wal_path(dir, 3), PathBuf::from("/tmp/wal/shard-3.wal")); - assert_eq!(wal_path(dir, 15), PathBuf::from("/tmp/wal/shard-15.wal")); - } - - #[test] - fn test_wal_replay_with_collections() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // HSET - let hset = make_command(&[b"HSET", b"myhash", b"f1", b"v1"]); - writer.append(&serialize_command(&hset)); - - // LPUSH - let lpush = make_command(&[b"LPUSH", b"mylist", b"a", b"b"]); - writer.append(&serialize_command(&lpush)); - - // SADD - let sadd = make_command(&[b"SADD", b"myset", b"x", b"y"]); - writer.append(&serialize_command(&sadd)); - - // ZADD - let zadd = make_command(&[b"ZADD", b"myzset", b"1.5", b"alice"]); - writer.append(&serialize_command(&zadd)); - - writer.flush_sync().unwrap(); - - // Replay - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, writer.path(), &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 4); - - // Verify hash - let hash = dbs[0].get_hash(b"myhash").unwrap().unwrap(); - assert_eq!( - hash.get(&Bytes::from_static(b"f1")).unwrap().as_ref(), - b"v1" - ); - - // Verify list - let list = dbs[0].get_list(b"mylist").unwrap().unwrap(); - assert_eq!(list.len(), 2); - - // Verify set - let set = dbs[0].get_set(b"myset").unwrap().unwrap(); - assert_eq!(set.len(), 2); - - // Verify sorted set - let (members, _) = dbs[0].get_sorted_set(b"myzset").unwrap().unwrap(); - assert_eq!(*members.get(&Bytes::from_static(b"alice")).unwrap(), 1.5); - } - - #[test] - fn test_wal_v2_header_format() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(3, dir.path()).unwrap(); - - // Append and flush one command to ensure file has content beyond header - let cmd = make_command(&[b"SET", b"x", b"y"]); - let data = serialize_to_bytes(&cmd); - writer.append(&data); - writer.flush_if_needed().unwrap(); - - let contents = std::fs::read(writer.path()).unwrap(); - assert!(contents.len() >= WAL_HEADER_SIZE); - - // Verify header fields - assert_eq!(&contents[0..6], b"RRDWAL", "magic bytes"); - assert_eq!(contents[6], 2, "version"); - let shard_id = u16::from_le_bytes(contents[7..9].try_into().unwrap()); - assert_eq!(shard_id, 3, "shard_id"); - let epoch = u64::from_le_bytes(contents[9..17].try_into().unwrap()); - assert_eq!(epoch, 0, "epoch"); - // Reserved bytes should all be zero - assert_eq!(&contents[17..32], &[0u8; 15], "reserved bytes"); - } - - #[test] - fn test_wal_v2_block_crc_valid() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - let cmd = make_command(&[b"SET", b"mykey", b"myval"]); - let data = serialize_to_bytes(&cmd); - writer.append(&data); - writer.flush_if_needed().unwrap(); - - let contents = std::fs::read(writer.path()).unwrap(); - - // Parse block after 32-byte header - let bs = WAL_HEADER_SIZE; - let block_len = u32::from_le_bytes(contents[bs..bs + 4].try_into().unwrap()); - let cmd_count_bytes = &contents[bs + 4..bs + 6]; - let cmd_count = u16::from_le_bytes(cmd_count_bytes.try_into().unwrap()); - assert_eq!(cmd_count, 1); - let db_idx = contents[bs + 6]; - assert_eq!(db_idx, 0); - - // Extract payload and stored CRC - let payload_len = block_len as usize - 2 - 1 - 4; - let payload = &contents[bs + 7..bs + 7 + payload_len]; - let crc_offset = bs + 7 + payload_len; - let stored_crc = - u32::from_le_bytes(contents[crc_offset..crc_offset + 4].try_into().unwrap()); - - // Recompute CRC over cmd_count + db_idx + payload - let mut hasher = crc32fast::Hasher::new(); - hasher.update(cmd_count_bytes); - hasher.update(&[db_idx]); - hasher.update(payload); - let computed_crc = hasher.finalize(); - - assert_eq!(stored_crc, computed_crc, "CRC32 checksum mismatch"); - - // Verify payload is the original RESP data - assert_eq!(payload, &data[..]); - } - - #[test] - fn test_wal_v1_backward_compat() { - // Write raw RESP bytes (v1 format -- no header, no framing) - let dir = tempdir().unwrap(); - let path = dir.path().join("shard-0.wal"); - - let cmd1 = make_command(&[b"SET", b"oldkey1", b"oldval1"]); - let cmd2 = make_command(&[b"SET", b"oldkey2", b"oldval2"]); - let mut raw = Vec::new(); - raw.extend_from_slice(&serialize_to_bytes(&cmd1)); - raw.extend_from_slice(&serialize_to_bytes(&cmd2)); - std::fs::write(&path, &raw).unwrap(); - - // replay_wal should auto-detect v1 (no RRDWAL magic) and delegate to replay_aof - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, &path, &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 2); - assert_eq!( - dbs[0].get(b"oldkey1").unwrap().value.as_bytes().unwrap(), - b"oldval1" - ); - assert_eq!( - dbs[0].get(b"oldkey2").unwrap().value.as_bytes().unwrap(), - b"oldval2" - ); - } - - #[test] - fn test_wal_v2_corruption_stops_replay() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Block 1: 3 SET commands - let cmd1 = make_command(&[b"SET", b"key1", b"val1"]); - let cmd2 = make_command(&[b"SET", b"key2", b"val2"]); - let cmd3 = make_command(&[b"SET", b"key3", b"val3"]); - writer.append(&serialize_command(&cmd1)); - writer.append(&serialize_command(&cmd2)); - writer.append(&serialize_command(&cmd3)); - writer.flush_if_needed().unwrap(); - - // Block 2: 2 SET commands - let cmd4 = make_command(&[b"SET", b"key4", b"val4"]); - let cmd5 = make_command(&[b"SET", b"key5", b"val5"]); - writer.append(&serialize_command(&cmd4)); - writer.append(&serialize_command(&cmd5)); - writer.flush_sync().unwrap(); - - // Read the file and corrupt the CRC of the second block - let mut data = std::fs::read(writer.path()).unwrap(); - - // Find second block: after header(32) + first block(4 + block_len) - let first_block_len = u32::from_le_bytes( - data[WAL_HEADER_SIZE..WAL_HEADER_SIZE + 4] - .try_into() - .unwrap(), - ) as usize; - let second_block_start = WAL_HEADER_SIZE + 4 + first_block_len; - let second_block_len = u32::from_le_bytes( - data[second_block_start..second_block_start + 4] - .try_into() - .unwrap(), - ) as usize; - - // Flip a byte in the CRC of the second block (last 4 bytes of block content) - let crc_offset = second_block_start + 4 + second_block_len - 4; - data[crc_offset] ^= 0xFF; - - // Write corrupted data to a new file - let corrupted_path = dir.path().join("corrupted.wal"); - std::fs::write(&corrupted_path, &data).unwrap(); - - // Replay should stop after first block (3 commands) - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, &corrupted_path, &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 3, "should replay only first block's 3 commands"); - - // Keys from block 1 should be set - assert_eq!( - dbs[0].get(b"key1").unwrap().value.as_bytes().unwrap(), - b"val1" - ); - assert_eq!( - dbs[0].get(b"key2").unwrap().value.as_bytes().unwrap(), - b"val2" - ); - assert_eq!( - dbs[0].get(b"key3").unwrap().value.as_bytes().unwrap(), - b"val3" - ); - - // Keys from block 2 should NOT be set (corruption stopped replay) - assert!(dbs[0].get(b"key4").is_none(), "key4 should not exist"); - assert!(dbs[0].get(b"key5").is_none(), "key5 should not exist"); - } - - #[test] - fn test_wal_v2_truncated_block_stops() { - let dir = tempdir().unwrap(); - let mut writer = WalWriter::new(0, dir.path()).unwrap(); - - // Write one valid block - let cmd = make_command(&[b"SET", b"tkey", b"tval"]); - writer.append(&serialize_command(&cmd)); - writer.flush_sync().unwrap(); - - // Read file and truncate mid-way through what would be a second block - let mut data = std::fs::read(writer.path()).unwrap(); - // Append a partial block: block_len says 100, but only provide 5 bytes of content - data.extend_from_slice(&100u32.to_le_bytes()); - data.extend_from_slice(&[0u8; 5]); - - let truncated_path = dir.path().join("truncated.wal"); - std::fs::write(&truncated_path, &data).unwrap(); - - // Replay should return the 1 command from the valid first block - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, &truncated_path, &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 1); - assert_eq!( - dbs[0].get(b"tkey").unwrap().value.as_bytes().unwrap(), - b"tval" - ); - } - - #[test] - fn test_wal_v2_empty_after_header() { - let dir = tempdir().unwrap(); - let _writer = WalWriter::new(0, dir.path()).unwrap(); - - // Writer creates file with just the 32-byte header (no appends, no flush) - let path = wal_path(dir.path(), 0); - let mut dbs = vec![Database::new()]; - let count = replay_wal(&mut dbs, &path, &DispatchReplayEngine::new()).unwrap(); - assert_eq!(count, 0); - } -} diff --git a/src/persistence/wal_v3/record.rs b/src/persistence/wal_v3/record.rs index 1d3a8cab..11520ac3 100644 --- a/src/persistence/wal_v3/record.rs +++ b/src/persistence/wal_v3/record.rs @@ -57,14 +57,15 @@ pub enum WalRecordType { FileTierChange = 0x42, /// Cross-store transaction begin. XactBegin = 0x50, - /// Cross-store transaction commit (contains all store operations). - XactCommit = 0x51, /// Cross-store transaction abort. XactAbort = 0x52, - /// Cross-store transaction commit, v2: header carries the logical db - /// index so crash replay restores KV ops into the db the transaction ran - /// in (v1 records replay into db 0 — the pre-multi-db behaviour). - XactCommitV2 = 0x53, + /// Cross-store transaction commit (contains all store operations). The + /// header carries the logical db index so crash replay restores KV ops + /// into the db the transaction ran in. Discriminant 0x53 is retained from + /// the pre-1.0 "v2" record (0x51 was the original db-0-hardcoded + /// encoding, deleted in the WAL-v3-only format freeze — 0x51 is not + /// reused). + XactCommit = 0x53, /// Workspace creation record. WorkspaceCreate = 0x60, /// Workspace deletion record. @@ -94,9 +95,8 @@ impl WalRecordType { 0x41 => Some(Self::FileDelete), 0x42 => Some(Self::FileTierChange), 0x50 => Some(Self::XactBegin), - 0x51 => Some(Self::XactCommit), 0x52 => Some(Self::XactAbort), - 0x53 => Some(Self::XactCommitV2), + 0x53 => Some(Self::XactCommit), 0x60 => Some(Self::WorkspaceCreate), 0x61 => Some(Self::WorkspaceDrop), 0x70 => Some(Self::MqCreate), @@ -313,13 +313,16 @@ pub fn decode_graph_temporal(payload: &[u8]) -> Option<(u64, bool, i64, i64)> { Some((entity_id, is_node, valid_to, system_from)) } -/// Encode XactCommit WAL payload for crash recovery. +/// Encode a [`WalRecordType::XactCommit`] payload for crash recovery. /// -/// Layout matches `replay_xact_commit()` decoder exactly: -/// `[txn_id: u64 LE][kv_op_count: u32 LE][ops...]` -/// Per op: `[op_type: u8 (0=SET, 1=DEL)]` -/// `[key_len: u32 LE][key bytes]` -/// `[value_len: u32 LE (SET only)][value bytes (SET only)]` +/// The header carries the logical db index the transaction ran in, so crash +/// replay restores the KV ops into the correct db (`replay_xact_commit` in +/// `wal_v3::replay`). +/// +/// Layout (little-endian): `txn_id: u64, db_index: u32, kv_op_count: u32, +/// ops…`. Per op: `[op_type: u8 (0=SET, 1=DEL)]` +/// `[key_len: u32 LE][key bytes]` +/// `[value_len: u32 LE (SET only)][value bytes (SET only)]`. /// /// For SET ops, reads the CURRENT value from `db.data()` (post-commit state). /// For DEL ops, encodes just the key (replay calls `db.remove()`). @@ -327,55 +330,6 @@ pub fn decode_graph_temporal(payload: &[u8]) -> Option<(u64, bool, i64, i64)> { /// The undo log carries before-images for rollback; the WAL needs /// forward-images (final committed state) for replay. pub fn encode_xact_commit_payload( - txn_id: u64, - undo_records: &[crate::transaction::UndoRecord], - db: &crate::storage::Database, -) -> Vec { - let mut payload = Vec::with_capacity(12 + undo_records.len() * 32); - payload.extend_from_slice(&txn_id.to_le_bytes()); - // Count placeholder — patched at the end - let count_offset = payload.len(); - payload.extend_from_slice(&0u32.to_le_bytes()); - let mut count = 0u32; - - for record in undo_records { - match record { - crate::transaction::UndoRecord::Insert { key } - | crate::transaction::UndoRecord::Update { key, .. } => { - // Read current (post-dispatch) value for forward-image WAL - if let Some(entry) = db.data().get(key.as_ref()) { - if let Some(value) = entry.value.as_bytes_owned() { - payload.push(0u8); // op_type = SET - payload.extend_from_slice(&(key.len() as u32).to_le_bytes()); - payload.extend_from_slice(key); - payload.extend_from_slice(&(value.len() as u32).to_le_bytes()); - payload.extend_from_slice(&value); - count += 1; - } - } - } - crate::transaction::UndoRecord::Delete { key, .. } => { - payload.push(1u8); // op_type = DEL - payload.extend_from_slice(&(key.len() as u32).to_le_bytes()); - payload.extend_from_slice(key); - count += 1; - } - } - } - // Patch count field - payload[count_offset..count_offset + 4].copy_from_slice(&count.to_le_bytes()); - payload -} - -/// Encode an [`WalRecordType::XactCommitV2`] payload. -/// -/// Identical to the v1 format except the header carries the logical db index -/// the transaction ran in, so crash replay restores the KV ops into the -/// correct db (v1 records hardcoded db 0 on replay). -/// -/// Layout (little-endian): `txn_id: u64, db_index: u32, kv_op_count: u32, -/// ops…` (op wire format unchanged from v1). -pub fn encode_xact_commit_payload_v2( txn_id: u64, db_index: u32, undo_records: &[crate::transaction::UndoRecord], @@ -495,8 +449,8 @@ mod tests { assert_eq!(WalRecordType::FileDelete as u8, 0x41); assert_eq!(WalRecordType::FileTierChange as u8, 0x42); assert_eq!(WalRecordType::XactBegin as u8, 0x50); - assert_eq!(WalRecordType::XactCommit as u8, 0x51); assert_eq!(WalRecordType::XactAbort as u8, 0x52); + assert_eq!(WalRecordType::XactCommit as u8, 0x53); assert_eq!(WalRecordType::WorkspaceCreate as u8, 0x60); assert_eq!(WalRecordType::WorkspaceDrop as u8, 0x61); assert_eq!(WalRecordType::MqCreate as u8, 0x70); @@ -505,11 +459,25 @@ mod tests { // from_u8 roundtrips for &v in &[ 0x01, 0x10, 0x20, 0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x40, 0x41, 0x42, 0x50, - 0x51, 0x52, 0x60, 0x61, 0x70, 0x71, + 0x52, 0x53, 0x60, 0x61, 0x70, 0x71, ] { assert!(WalRecordType::from_u8(v).is_some()); } assert!(WalRecordType::from_u8(0xFF).is_none()); + // 0x51 was the deleted pre-1.0 v1 XactCommit discriminant. It is + // deliberately NOT reused -- a v3 stream containing a stray 0x51 + // record (e.g. written by a pre-freeze build) must decode as an + // unrecognized tag, not silently reinterpreted as today's + // XactCommit. Note the replay engine does NOT skip just this one + // record: `read_wal_v3_record` returns `None` for any unrecognized + // tag, and the segment replay loop (`replay_wal_v3_file_until` in + // `wal_v3::replay`) treats `None` identically to a corrupt/truncated + // record -- it warns and stops replaying the REST of that segment. + // (0x51 was already dead in production before this refactor -- no + // shipped build ever wrote it via a real TXN.COMMIT -- so this is a + // latent robustness note for future record-type additions, not a + // live data-loss path today.) + assert!(WalRecordType::from_u8(0x51).is_none()); } #[test] @@ -656,14 +624,15 @@ mod tests { }, ]; - let payload = encode_xact_commit_payload(42, &records, &db); + let payload = encode_xact_commit_payload(42, 3, &records, &db); // Verify header assert_eq!(u64::from_le_bytes(payload[0..8].try_into().unwrap()), 42); // txn_id - assert_eq!(u32::from_le_bytes(payload[8..12].try_into().unwrap()), 2); // 2 ops + assert_eq!(u32::from_le_bytes(payload[8..12].try_into().unwrap()), 3); // db_index + assert_eq!(u32::from_le_bytes(payload[12..16].try_into().unwrap()), 2); // 2 ops // Verify op 1: SET key1 committed_val - let mut offset = 12; + let mut offset = 16; assert_eq!(payload[offset], 0); // SET offset += 1; let key_len = u32::from_le_bytes(payload[offset..offset + 4].try_into().unwrap()) as usize; diff --git a/src/persistence/wal_v3/replay.rs b/src/persistence/wal_v3/replay.rs index 2cdb339b..2fcb6783 100644 --- a/src/persistence/wal_v3/replay.rs +++ b/src/persistence/wal_v3/replay.rs @@ -1,9 +1,11 @@ -//! WAL v3 replay engine — v2/v3 auto-detection, LSN-based skip, FPI callback. +//! WAL v3 replay engine — v3 is the only WAL format; auto-detection +//! distinguishes it from raw AOF and rejects the removed v2 format. //! //! The replay engine is the recovery path after crash or restart. It handles: -//! - v2 WAL files (version byte=2) by delegating to the existing v2 replay path //! - v3 WAL files (version byte=3) with per-record LSN tracking -//! - Raw RESP (v1) by delegating to AOF replay +//! - Raw RESP (v1 AOF) by delegating to AOF replay +//! - v2 WAL files (version byte=2) are rejected loudly: WAL v2 was removed in +//! the pre-1.0 WAL-v3-only format freeze (see `WalError::UnsupportedVersion`) //! - Auto-detection at byte offset 6 to distinguish formats //! //! FPI (Full Page Image) records during replay unconditionally overwrite the @@ -31,10 +33,11 @@ pub struct WalV3ReplayResult { /// Auto-detect WAL format and replay accordingly. /// /// Reads the first bytes of the file to determine the format: -/// - `RRDWAL` magic + version=2 => delegate to existing v2 replay /// - `RRDWAL` magic + version=3 => use v3 replay engine /// - No `RRDWAL` magic => delegate to AOF (raw RESP v1) replay -/// - Other version => return UnsupportedVersion error +/// - `RRDWAL` magic + version=2 => return `UnsupportedVersion` (WAL v2 was +/// removed in the pre-1.0 WAL-v3-only format freeze; re-ingest via AOF) +/// - Any other version => return `UnsupportedVersion` error pub fn replay_wal_auto( databases: &mut [crate::storage::Database], path: &Path, @@ -49,8 +52,18 @@ pub fn replay_wal_auto( if data.len() >= WAL_V3_HEADER_SIZE && data[..6] == *WAL_V3_MAGIC { match data[6] { 2 => { - // v2 format — delegate to existing replay - crate::persistence::wal::replay_wal(databases, path, engine) + // WAL v2 was removed in the pre-1.0 WAL-v3-only format + // freeze -- this build cannot replay it. Fail loudly rather + // than silently dropping writes; the operator must re-ingest + // via the AOF (the recovery authority) or re-import from a + // pre-freeze build. + tracing::error!( + "WAL v2 file {:?} found but WAL v2 support was removed \ + (pre-1.0 WAL-v3-only format freeze) — re-ingest via AOF \ + or replay it on a pre-freeze Moon build first", + path + ); + Err(crate::error::WalError::UnsupportedVersion { version: 2 }.into()) } 3 => { // v3 format — replay commands through engine @@ -73,15 +86,10 @@ pub fn replay_wal_auto( tracing::trace!(lsn = record.lsn, "WAL replay: XactBegin"); } WalRecordType::XactCommit => { - // XactCommit: replay KV ops from payload + // XactCommit: db-index-aware KV replay replay_xact_commit(databases, &record.payload); tracing::trace!(lsn = record.lsn, "WAL replay: XactCommit"); } - WalRecordType::XactCommitV2 => { - // XactCommitV2: db-index-aware KV replay - replay_xact_commit_v2(databases, &record.payload); - tracing::trace!(lsn = record.lsn, "WAL replay: XactCommitV2"); - } WalRecordType::XactAbort => { // XactAbort: no action - changes were never committed tracing::trace!(lsn = record.lsn, "WAL replay: XactAbort"); @@ -156,6 +164,90 @@ pub fn replay_wal_auto( } } +/// Replay every Command/XactCommit record across all segments in a WAL v3 +/// directory through `engine`, applying the same record-type dispatch as +/// `replay_wal_auto`'s v3 branch (this function is the multi-segment, +/// directory-scoped counterpart — `replay_wal_auto` only handles one file). +/// +/// Used by legacy (non-disk-offload) shard recovery: WAL v3 is written even +/// when `--disk-offload` is off (rooted at `/shard-N/wal-v3`, +/// see `event_loop::run`'s writer-creation block), fsynced on a 1s cadence. +/// +/// ⚠ NOT a recovery authority. Post-#211 WAL v3's KV coverage is +/// intentionally PARTIAL: `--wal-kv-log` is default-off and connection-local +/// KV writes bypass WAL v3 entirely even when it is on (measured 79.2% +/// incomplete WAL-only recovery, see `tmp/WRITE-DIAG.md`). The retired +/// WAL v2 "prefer the WAL, skip the AOF" contract must NOT be applied to +/// this helper — a non-empty return typically counts temporal/graph/txn +/// records while ALL local KV data lives only in the AOF. Callers replay +/// `appendonly.aof` as the authority and use this helper ONLY as a +/// last-resort fallback when no AOF exists (partial recovery beats none), +/// logging a loud partial-coverage warning when they do. +/// +/// Returns `Ok(0)` (not an error) when `wal_dir` does not exist or is empty. +pub fn replay_wal_v3_dir_commands( + wal_dir: &Path, + databases: &mut [crate::storage::Database], + engine: &dyn crate::persistence::replay::CommandReplayEngine, +) -> std::io::Result { + if !wal_dir.exists() { + return Ok(0); + } + + let mut commands_replayed = 0usize; + let mut selected_db = 0usize; + let on_command = &mut |record: &WalRecord| { + match record.record_type { + WalRecordType::Command => { + engine.replay_command(databases, &record.payload, &[], &mut selected_db); + } + WalRecordType::XactBegin => { + tracing::trace!(lsn = record.lsn, "WAL replay: XactBegin"); + } + WalRecordType::XactCommit => { + replay_xact_commit(databases, &record.payload); + tracing::trace!(lsn = record.lsn, "WAL replay: XactCommit"); + } + WalRecordType::XactAbort => { + tracing::trace!(lsn = record.lsn, "WAL replay: XactAbort"); + } + WalRecordType::TemporalUpsert => { + if decode_temporal_upsert(&record.payload).is_none() { + tracing::warn!( + lsn = record.lsn, + "WAL replay: malformed TemporalUpsert payload" + ); + } + // Full state restoration happens in the temporal replay path + // (TemporalKvIndex lives on ShardDatabases, not on databases). + } + WalRecordType::GraphTemporal => { + if decode_graph_temporal(&record.payload).is_none() { + tracing::warn!( + lsn = record.lsn, + "WAL replay: malformed GraphTemporal payload" + ); + } + // Full state restoration happens in the graph replay path + // (GraphStore lives on ShardDatabases, not on databases). + } + _ => { + // Other record types (Vector*, Checkpoint, etc.) are not + // meaningful for the legacy KV-only replay path. + } + } + commands_replayed += 1; + }; + let on_fpi = &mut |_record: &WalRecord| { + // No page cache in legacy (non-disk-offload) mode; FPI records are + // never written there (WalWriterV3 is created without a page-cache + // handle), so this is unreachable in practice but kept for symmetry. + }; + + replay_wal_v3_dir(wal_dir, 0, on_command, on_fpi)?; + Ok(commands_replayed) +} + /// Replay all WAL v3 segment files in a directory. /// /// Scans `wal_dir` for `*.wal` files, sorts by filename (zero-padded sequence @@ -315,7 +407,6 @@ pub fn replay_wal_v3_file_until( | WalRecordType::FileTierChange | WalRecordType::XactBegin | WalRecordType::XactCommit - | WalRecordType::XactCommitV2 | WalRecordType::XactAbort | WalRecordType::TemporalUpsert | WalRecordType::GraphTemporal @@ -423,49 +514,24 @@ pub fn resolve_target_time_to_lsn( Ok(best) } -/// Replay a cross-store transaction commit record. +/// Replay a cross-store transaction commit record (db-index-aware). /// -/// The XactCommit payload format (little-endian): -/// - txn_id: u64 -/// - kv_op_count: u32 -/// - For each KV op: +/// Payload layout (little-endian): `txn_id: u64, db_index: u32, +/// kv_op_count: u32, ops…`. Per op: /// - op_type: u8 (0=SET, 1=DEL) /// - key_len: u32 /// - key: [u8; key_len] /// - value_len: u32 (only for SET) /// - value: [u8; value_len] (only for SET) /// +/// An out-of-range `db_index` (server restarted with fewer `--databases`) +/// falls back to db 0 with a loud warning rather than dropping the ops. +/// /// Vector and graph ops are handled by their respective replay paths /// (VectorTxnCommit already exists). fn replay_xact_commit(databases: &mut [crate::storage::Database], payload: &[u8]) { - if payload.len() < 12 { - tracing::warn!("XactCommit payload too short: {} bytes", payload.len()); - return; - } - - let txn_id = u64::from_le_bytes([ - payload[0], payload[1], payload[2], payload[3], payload[4], payload[5], payload[6], - payload[7], - ]); - let kv_op_count = - u32::from_le_bytes([payload[8], payload[9], payload[10], payload[11]]) as usize; - - tracing::debug!(txn_id, kv_op_count, "Replaying XactCommit"); - - // v1 records predate the db-index header: they were only ever written by - // builds that replayed into db 0, so db 0 is the faithful target. - replay_xact_kv_ops(&mut databases[0], payload, 12, kv_op_count); -} - -/// Replay a cross-store transaction commit record, v2 (db-index-aware). -/// -/// Payload layout (little-endian): `txn_id: u64, db_index: u32, -/// kv_op_count: u32, ops…` — op wire format identical to v1. -/// An out-of-range `db_index` (server restarted with fewer `--databases`) -/// falls back to db 0 with a loud warning rather than dropping the ops. -fn replay_xact_commit_v2(databases: &mut [crate::storage::Database], payload: &[u8]) { if payload.len() < 16 { - tracing::warn!("XactCommitV2 payload too short: {} bytes", payload.len()); + tracing::warn!("XactCommit payload too short: {} bytes", payload.len()); return; } @@ -477,7 +543,7 @@ fn replay_xact_commit_v2(databases: &mut [crate::storage::Database], payload: &[ let kv_op_count = u32::from_le_bytes([payload[12], payload[13], payload[14], payload[15]]) as usize; - tracing::debug!(txn_id, db_index, kv_op_count, "Replaying XactCommitV2"); + tracing::debug!(txn_id, db_index, kv_op_count, "Replaying XactCommit"); let target = if db_index < databases.len() { db_index @@ -486,7 +552,7 @@ fn replay_xact_commit_v2(databases: &mut [crate::storage::Database], payload: &[ txn_id, db_index, db_count = databases.len(), - "XactCommitV2 db index out of range (server restarted with fewer \ + "XactCommit db index out of range (server restarted with fewer \ --databases?) — replaying into db 0" ); 0 @@ -494,7 +560,7 @@ fn replay_xact_commit_v2(databases: &mut [crate::storage::Database], payload: &[ replay_xact_kv_ops(&mut databases[target], payload, 16, kv_op_count); } -/// Shared op-loop for XactCommit v1/v2: apply `kv_op_count` SET/DEL ops read +/// Shared op-loop for XactCommit: apply `kv_op_count` SET/DEL ops read /// from `payload` starting at `offset` to `db`. Truncation-defensive — a /// short payload warns and stops rather than panicking. fn replay_xact_kv_ops( @@ -591,12 +657,12 @@ mod tests { header } - // D-1: XactCommitV2 must replay KV ops into the db the transaction ran - // in — the v1 path hardcoded db 0, silently corrupting recovery for any - // TXN.BEGIN…COMMIT issued under SELECT != 0. + // D-1: XactCommit must replay KV ops into the db the transaction ran + // in — the deleted pre-1.0 v1 record hardcoded db 0, silently corrupting + // recovery for any TXN.BEGIN…COMMIT issued under SELECT != 0. #[test] - fn test_xact_commit_v2_replays_into_selected_db() { - use crate::persistence::wal_v3::record::encode_xact_commit_payload_v2; + fn test_xact_commit_replays_into_selected_db() { + use crate::persistence::wal_v3::record::encode_xact_commit_payload; use crate::storage::Database; use crate::transaction::UndoRecord; use bytes::Bytes; @@ -607,14 +673,14 @@ mod tests { let records = vec![UndoRecord::Insert { key: Bytes::from_static(b"txnkey"), }]; - let payload = encode_xact_commit_payload_v2(7, 3, &records, &src_db); + let payload = encode_xact_commit_payload(7, 3, &records, &src_db); let mut databases: Vec = (0..4).map(|_| Database::new()).collect(); - replay_xact_commit_v2(&mut databases, &payload); + replay_xact_commit(&mut databases, &payload); assert!( databases[3].data().get(b"txnkey".as_ref()).is_some(), - "XactCommitV2 must restore the key into db 3" + "XactCommit must restore the key into db 3" ); assert!( databases[0].data().get(b"txnkey".as_ref()).is_none(), @@ -623,8 +689,8 @@ mod tests { } #[test] - fn test_xact_commit_v2_out_of_range_db_falls_back_to_db0() { - use crate::persistence::wal_v3::record::encode_xact_commit_payload_v2; + fn test_xact_commit_out_of_range_db_falls_back_to_db0() { + use crate::persistence::wal_v3::record::encode_xact_commit_payload; use crate::storage::Database; use crate::transaction::UndoRecord; use bytes::Bytes; @@ -635,10 +701,10 @@ mod tests { key: Bytes::from_static(b"k"), }]; // db_index 9 but the restarted server only has 2 dbs. - let payload = encode_xact_commit_payload_v2(8, 9, &records, &src_db); + let payload = encode_xact_commit_payload(8, 9, &records, &src_db); let mut databases: Vec = (0..2).map(|_| Database::new()).collect(); - replay_xact_commit_v2(&mut databases, &payload); + replay_xact_commit(&mut databases, &payload); assert!( databases[0].data().get(b"k".as_ref()).is_some(), diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index a2f4b382..2ab9ad5b 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/src/server/conn/handler_monoio/txn.rs @@ -87,13 +87,13 @@ pub(super) async fn try_handle_txn_commit( return true; } - // Write XactCommitV2 WAL record with committed KV state. + // Write XactCommit WAL record with committed KV state. // Both the forward-image read and the record header use the // txn's BEGIN-time db so commit and replay agree on the db. let txn_id = txn.txn_id; if !txn.kv_undo.is_empty() { let payload = crate::shard::slice::with_shard_db(txn.db_index, |db| { - crate::persistence::wal_v3::record::encode_xact_commit_payload_v2( + crate::persistence::wal_v3::record::encode_xact_commit_payload( txn_id, txn.db_index as u32, txn.kv_undo.records(), @@ -104,7 +104,7 @@ pub(super) async fn try_handle_txn_commit( crate::persistence::wal_v3::record::write_wal_v3_record( &mut wal_buf, txn_id, - crate::persistence::wal_v3::record::WalRecordType::XactCommitV2, + crate::persistence::wal_v3::record::WalRecordType::XactCommit, &payload, ); ctx.shard_databases diff --git a/src/server/conn/handler_sharded/txn.rs b/src/server/conn/handler_sharded/txn.rs index cac87e1d..cbdd2eeb 100644 --- a/src/server/conn/handler_sharded/txn.rs +++ b/src/server/conn/handler_sharded/txn.rs @@ -87,14 +87,14 @@ pub(super) async fn try_handle_txn_commit( return true; } - // Write XactCommitV2 WAL record with committed KV state. + // Write XactCommit WAL record with committed KV state. // Both the forward-image read and the record header use the // txn's BEGIN-time db so commit and replay agree on the db. let txn_id = txn.txn_id; if !txn.kv_undo.is_empty() { // Unconditional slice path: ShardSlice is always initialized. let payload = crate::shard::slice::with_shard_db(txn.db_index, |db| { - crate::persistence::wal_v3::record::encode_xact_commit_payload_v2( + crate::persistence::wal_v3::record::encode_xact_commit_payload( txn_id, txn.db_index as u32, txn.kv_undo.records(), @@ -105,7 +105,7 @@ pub(super) async fn try_handle_txn_commit( crate::persistence::wal_v3::record::write_wal_v3_record( &mut wal_buf, txn_id, - crate::persistence::wal_v3::record::WalRecordType::XactCommitV2, + crate::persistence::wal_v3::record::WalRecordType::XactCommit, &payload, ); ctx.shard_databases diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 79b8b5d0..27857426 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -17,7 +17,6 @@ use crate::config::RuntimeConfig; use crate::persistence::control::ShardControlFile; use crate::persistence::page_cache::PageCache; use crate::persistence::snapshot::SnapshotState; -use crate::persistence::wal::WalWriter; use crate::persistence::wal_v3::segment::WalWriterV3; use crate::pubsub::PubSubRegistry; use crate::replication::state::ReplicationState; @@ -389,10 +388,14 @@ impl super::Shard { let mut snapshot_state: Option = None; let mut snapshot_reply_tx: Option>> = None; - // Per-shard WAL v2 writer — created only when persistence is enabled AND - // disk-offload is disabled. When disk-offload is on, WAL v3 supersedes v2 - // (stronger guarantees: per-record LSN + CRC32C), so v2 is skipped entirely - // to avoid double-write overhead. + // Per-shard WAL v3 writer — the only WAL (pre-1.0 format freeze dropped + // WAL v2). Created whenever persistence is enabled (`--appendonly yes`), + // rooted at: + // - `/shard-N/wal-v3` when disk-offload is enabled + // (matches the page-cache/manifest/control-file layout), or + // - `/shard-N/wal-v3` otherwise — the old WAL v2 + // niche, now served by the v3 segment format so WAL durability/CDC + // parity is preserved when disk-offload is off. let appendonly_enabled = runtime_config.read().appendonly != "no"; // `--wal-kv-log`: whether SPSC-executed KV writes are also logged to // the per-shard WAL. Resolved per drain cycle (Auto is dynamic on the @@ -404,63 +407,43 @@ impl super::Shard { "--wal-kv-log off with --appendonly no: KV writes have NO durability log" ); } - let mut wal_writer: Option = match (&persistence_dir, appendonly_enabled) { - (Some(dir), true) if !server_config.disk_offload_enabled() => { - match WalWriter::new(shard_id, std::path::Path::new(dir)) { - Ok(w) => { - info!("Shard {}: WAL writer initialized", shard_id); - Some(w) - } - Err(e) => { - tracing::warn!("Shard {}: WAL init failed: {}", shard_id, e); - None - } - } - } - (Some(_), true) => { - info!( - "Shard {}: WAL v2 skipped (disk-offload active, WAL v3 supersedes)", - shard_id - ); - None - } - (Some(_), false) => { - info!("Shard {}: WAL skipped (appendonly=no)", shard_id); - None - } - (None, _) => None, - }; // Disk-offload base directory (None when disk-offload is disabled). let disk_offload_base: Option = server_config .disk_offload_enabled() .then(|| server_config.effective_disk_offload_dir()); - // Per-shard WAL v3 writer (created only when disk-offload is enabled). - // Provides per-record LSN tracking and FPI support for checkpoint-based recovery. - // WAL v2 remains active for non-disk-offload mode; both writers can coexist. - let mut wal_v3_writer: Option = if server_config.disk_offload_enabled() { - let shard_dir = server_config - .effective_disk_offload_dir() - .join(format!("shard-{}", shard_id)); + let wal_shard_dir: Option = if !appendonly_enabled { + info!("Shard {}: WAL skipped (appendonly=no)", shard_id); + None + } else if server_config.disk_offload_enabled() { + Some( + server_config + .effective_disk_offload_dir() + .join(format!("shard-{}", shard_id)), + ) + } else { + persistence_dir + .as_ref() + .map(|dir| std::path::Path::new(dir).join(format!("shard-{}", shard_id))) + }; + let mut wal_writer: Option = wal_shard_dir.and_then(|shard_dir| { let wal_dir = shard_dir.join("wal-v3"); match WalWriterV3::new(shard_id, &wal_dir, server_config.wal_segment_size_bytes()) { Ok(w) => { info!( - "Shard {}: WAL v3 writer initialized (segment_size={})", + "Shard {}: WAL writer initialized (segment_size={})", shard_id, server_config.wal_segment_size_bytes() ); Some(w) } Err(e) => { - tracing::warn!("Shard {}: WAL v3 init failed: {}", shard_id, e); + tracing::warn!("Shard {}: WAL init failed: {}", shard_id, e); None } } - } else { - None - }; + }); // Per-shard WAL append channel for local writes. // Connection handlers send serialized write commands here; we drain on the 1ms tick. @@ -1201,7 +1184,7 @@ impl super::Shard { let hit_cap = spsc_handler::drain_spsc_shared( &shard_databases, &mut consumers, &mut *pubsub_arc.write(), &blocking_rc, &mut pending_snapshot, &mut snapshot_state, - &mut wal_writer, &mut wal_v3_writer, &repl_backlog, &mut replica_txs, + &mut wal_writer, &repl_backlog, &mut replica_txs, &repl_offsets, shard_id, &script_cache_rc, &cached_clock, &mut pending_migrations, &mut pending_cdc_subscribes, @@ -1242,7 +1225,7 @@ impl super::Shard { } } if !pending_cdc_subscribes.is_empty() { - let wal_dir = wal_v3_writer.as_ref().map(|w| w.wal_dir()); + let wal_dir = wal_writer.as_ref().map(|w| w.wal_dir()); cdc_registry.register_pending( pending_cdc_subscribes.drain(..), wal_dir, ); @@ -1250,7 +1233,7 @@ impl super::Shard { persistence_tick::handle_pending_snapshot( pending_snapshot, &mut snapshot_state, &mut snapshot_reply_tx, &shard_databases, disk_offload_base.as_deref(), shard_id, - wal_v3_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), + wal_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), ); for (fd, state) in pending_migrations.drain(..) { #[cfg(unix)] @@ -1307,7 +1290,7 @@ impl super::Shard { let hit_cap = spsc_handler::drain_spsc_shared( &shard_databases, &mut consumers, &mut *pubsub_arc.write(), &blocking_rc, &mut pending_snapshot, &mut snapshot_state, - &mut wal_writer, &mut wal_v3_writer, &repl_backlog, &mut replica_txs, + &mut wal_writer, &repl_backlog, &mut replica_txs, &repl_offsets, shard_id, &script_cache_rc, &cached_clock, &mut pending_migrations, &mut pending_cdc_subscribes, @@ -1348,7 +1331,7 @@ impl super::Shard { } } if !pending_cdc_subscribes.is_empty() { - let wal_dir = wal_v3_writer.as_ref().map(|w| w.wal_dir()); + let wal_dir = wal_writer.as_ref().map(|w| w.wal_dir()); cdc_registry.register_pending( pending_cdc_subscribes.drain(..), wal_dir, ); @@ -1356,7 +1339,7 @@ impl super::Shard { persistence_tick::handle_pending_snapshot( pending_snapshot, &mut snapshot_state, &mut snapshot_reply_tx, &shard_databases, disk_offload_base.as_deref(), shard_id, - wal_v3_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), + wal_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), ); for (fd, state) in pending_migrations.drain(..) { #[cfg(unix)] @@ -1406,7 +1389,7 @@ impl super::Shard { &snapshot_trigger_rx, &mut last_snapshot_epoch, &mut snapshot_state, &shard_databases, &persistence_dir, disk_offload_base.as_deref(), shard_id, - wal_v3_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), + wal_writer.as_ref().map(|w| w.current_lsn().saturating_sub(1)).unwrap_or(0), ); // Advance snapshot one segment per tick (cooperative) @@ -1423,8 +1406,7 @@ impl super::Shard { ); } else { persistence_tick::finalize_snapshot_success( - &mut snapshot_state, &mut snapshot_reply_tx, - &mut wal_writer, shard_id, + &mut snapshot_state, &mut snapshot_reply_tx, shard_id, ); bgsave_checkpoint_requested = true; } @@ -1434,9 +1416,6 @@ impl super::Shard { // Drain local-write WAL channel (connection handler inline writes) while let Ok(data) = wal_append_rx.try_recv() { if let Some(ref mut wal) = wal_writer { - wal.append(&data); - } - if let Some(ref mut wal) = wal_v3_writer { wal.append( crate::persistence::wal_v3::record::WalRecordType::Command, &data, @@ -1444,8 +1423,7 @@ impl super::Shard { } } - persistence_tick::flush_wal_if_needed(&mut wal_writer); - persistence_tick::flush_wal_v3_if_needed(&mut wal_v3_writer); + persistence_tick::flush_wal_v3_if_needed(&mut wal_writer); // C3b-2 — Drive CDC fan-out AFTER flush_if_needed so the // tail reader sees the bytes just handed to the page @@ -1454,18 +1432,18 @@ impl super::Shard { cdc_registry.fanout_tick(cached_clock.ms() as i64); } - // appendfsync=always: fsync WAL v3 after every SPSC drain batch + // appendfsync=always: fsync WAL after every SPSC drain batch if server_config.appendfsync == "always" { - if let Some(ref mut wal) = wal_v3_writer { + if let Some(ref mut wal) = wal_writer { if let Err(e) = wal.flush_sync() { - tracing::error!("WAL v3 appendfsync=always failed: {}", e); + tracing::error!("WAL appendfsync=always failed: {}", e); } } } // Checkpoint protocol tick (disk-offload only) if let (Some(ckpt_mgr), Some(page_cache_inst), Some(wal_v3), Some(manifest), Some(ctrl), Some(ctrl_path)) = - (&mut checkpoint_manager, &page_cache, &mut wal_v3_writer, &mut shard_manifest, &mut control_file, &control_file_path) + (&mut checkpoint_manager, &page_cache, &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path) { // BGSAVE-triggered forced checkpoint (bypasses trigger conditions) if bgsave_checkpoint_requested && !ckpt_mgr.is_active() { @@ -1501,8 +1479,7 @@ impl super::Shard { } // WAL fsync + MVCC sweep on 1-second interval _ = wal_sync_interval.0.tick() => { - timers::sync_wal(&mut wal_writer); - timers::sync_wal_v3(&mut wal_v3_writer); + timers::sync_wal_v3(&mut wal_writer); // P3+MA1+MA2: prune committed + sweep zombies + kill old snapshots // + update RECL_MVCC_* + segment-stall. crate::shard::slice::with_shard(|s| { @@ -1518,7 +1495,7 @@ impl super::Shard { // P6: ceiling-trigger — runs at 1s cadence to avoid the // read_dir syscall overhead of wal.stats() on every 1ms tick. if let (Some(ckpt_mgr), Some(page_cache_inst), Some(wal_v3), Some(manifest), Some(ctrl), Some(ctrl_path)) = - (&mut checkpoint_manager, &page_cache, &mut wal_v3_writer, &mut shard_manifest, &mut control_file, &control_file_path) + (&mut checkpoint_manager, &page_cache, &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path) { if persistence_tick::maybe_force_checkpoint_on_wal_overflow( ckpt_mgr, @@ -1550,7 +1527,7 @@ impl super::Shard { server_config.segment_warm_after, &mut next_file_id, shard_id, - &mut wal_v3_writer, + &mut wal_writer, ); }); } @@ -1631,7 +1608,7 @@ impl super::Shard { &runtime_config, &page_cache, &mut next_file_id, - &mut wal_v3_writer, + &mut wal_writer, &script_cache_rc, &spill_file_id, ); @@ -1660,7 +1637,7 @@ impl super::Shard { #[cfg(feature = "graph")] &mut s.graph_store, shard_manifest.as_mut(), - wal_v3_writer.as_mut(), + wal_writer.as_mut(), server_config.max_wal_size_bytes(), server_config.manifest_tombstone_retain_epochs, server_config.manifest_tombstone_retain_secs, @@ -1681,7 +1658,7 @@ impl super::Shard { ); // Trigger final checkpoint before shutdown (design S9) if let (Some(ckpt_mgr), Some(page_cache_inst), Some(wal_v3), Some(manifest), Some(ctrl), Some(ctrl_path)) = - (&mut checkpoint_manager, &page_cache, &mut wal_v3_writer, &mut shard_manifest, &mut control_file, &control_file_path) + (&mut checkpoint_manager, &page_cache, &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path) { persistence_tick::force_checkpoint(ckpt_mgr, page_cache_inst, wal_v3, manifest, ctrl, ctrl_path, shard_id, server_config.manifest_tombstone_retain_epochs, server_config.manifest_tombstone_retain_secs); } @@ -1705,10 +1682,7 @@ impl super::Shard { }); } if let Some(ref mut wal) = wal_writer { - let _ = wal.shutdown(); - } - if let Some(ref mut wal_v3) = wal_v3_writer { - let _ = wal_v3.flush_sync(); + let _ = wal.flush_sync(); } break; } @@ -1826,7 +1800,7 @@ impl super::Shard { ) = ( &mut checkpoint_manager, &page_cache, - &mut wal_v3_writer, + &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path, @@ -1844,10 +1818,7 @@ impl super::Shard { ); } if let Some(ref mut wal) = wal_writer { - let _ = wal.shutdown(); - } - if let Some(ref mut wal_v3) = wal_v3_writer { - let _ = wal_v3.flush_sync(); + let _ = wal.flush_sync(); } break; } @@ -1887,7 +1858,6 @@ impl super::Shard { &mut pending_snapshot, &mut snapshot_state, &mut wal_writer, - &mut wal_v3_writer, &repl_backlog, &mut replica_txs, &repl_offsets, @@ -1922,7 +1892,7 @@ impl super::Shard { crate::admin::metrics_setup::bump_spsc_drain_renotify(); } if !pending_cdc_subscribes.is_empty() { - let wal_dir = wal_v3_writer.as_ref().map(|w| w.wal_dir()); + let wal_dir = wal_writer.as_ref().map(|w| w.wal_dir()); cdc_registry.register_pending(pending_cdc_subscribes.drain(..), wal_dir); } for waker in pending_wakers.borrow_mut().drain(..) { @@ -1935,7 +1905,7 @@ impl super::Shard { &shard_databases, disk_offload_base.as_deref(), shard_id, - wal_v3_writer + wal_writer .as_ref() .map(|w| w.current_lsn().saturating_sub(1)) .unwrap_or(0), @@ -2012,7 +1982,7 @@ impl super::Shard { &persistence_dir, disk_offload_base.as_deref(), shard_id, - wal_v3_writer + wal_writer .as_ref() .map(|w| w.current_lsn().saturating_sub(1)) .unwrap_or(0), @@ -2036,7 +2006,6 @@ impl super::Shard { persistence_tick::finalize_snapshot_success( &mut snapshot_state, &mut snapshot_reply_tx, - &mut wal_writer, shard_id, ); crate::command::persistence::bgsave_shard_done(true); @@ -2048,9 +2017,6 @@ impl super::Shard { // Drain local-write WAL channel while let Ok(data) = wal_append_rx.try_recv() { if let Some(ref mut wal) = wal_writer { - wal.append(&data); - } - if let Some(ref mut wal) = wal_v3_writer { wal.append( crate::persistence::wal_v3::record::WalRecordType::Command, &data, @@ -2058,8 +2024,7 @@ impl super::Shard { } } - persistence_tick::flush_wal_if_needed(&mut wal_writer); - persistence_tick::flush_wal_v3_if_needed(&mut wal_v3_writer); + persistence_tick::flush_wal_v3_if_needed(&mut wal_writer); // C3b-2 — Drive CDC fan-out on the same monoio tick body // as the tokio branch above. Zero CPU when no subscribers. @@ -2068,9 +2033,9 @@ impl super::Shard { } if server_config.appendfsync == "always" { - if let Some(ref mut wal) = wal_v3_writer { + if let Some(ref mut wal) = wal_writer { if let Err(e) = wal.flush_sync() { - tracing::error!("WAL v3 appendfsync=always failed: {}", e); + tracing::error!("WAL appendfsync=always failed: {}", e); } } } @@ -2086,7 +2051,7 @@ impl super::Shard { ) = ( &mut checkpoint_manager, &page_cache, - &mut wal_v3_writer, + &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path, @@ -2135,7 +2100,7 @@ impl super::Shard { &runtime_config, &page_cache, &mut next_file_id, - &mut wal_v3_writer, + &mut wal_writer, &script_cache_rc, &spill_file_id, ); @@ -2151,8 +2116,7 @@ impl super::Shard { // P6 is gated here (not per-1ms tick) to avoid the read_dir // syscall overhead of wal.stats() on the hot path. if monoio_tick_counter % 1000 == 0 { - timers::sync_wal(&mut wal_writer); - timers::sync_wal_v3(&mut wal_v3_writer); + timers::sync_wal_v3(&mut wal_writer); // P3+MA1+MA2: MVCC committed prune + zombie sweep + kill old snapshots // + RECL_* + segment-stall. crate::shard::slice::with_shard(|s| { @@ -2175,7 +2139,7 @@ impl super::Shard { ) = ( &mut checkpoint_manager, &page_cache, - &mut wal_v3_writer, + &mut wal_writer, &mut shard_manifest, &mut control_file, &control_file_path, @@ -2211,7 +2175,7 @@ impl super::Shard { server_config.segment_warm_after, &mut next_file_id, shard_id, - &mut wal_v3_writer, + &mut wal_writer, ); }); } @@ -2262,7 +2226,7 @@ impl super::Shard { #[cfg(feature = "graph")] &mut s.graph_store, shard_manifest.as_mut(), - wal_v3_writer.as_mut(), + wal_writer.as_mut(), server_config.max_wal_size_bytes(), server_config.manifest_tombstone_retain_epochs, server_config.manifest_tombstone_retain_secs, diff --git a/src/shard/mod.rs b/src/shard/mod.rs index 0fca0ce6..5d582654 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -122,13 +122,17 @@ impl Shard { /// Restore shard state from per-shard snapshot and WAL files at startup. /// /// When `disk_offload_dir` is `Some`, uses the v3 recovery protocol - /// (6-phase: control file -> manifest -> data load -> WAL v3 replay -> - /// consistency -> ready). Falls back to v2 path on v3 failure. + /// (6-phase: control file -> manifest -> data load -> WAL replay -> + /// consistency -> ready). Falls back to the legacy path on v3 failure. /// - /// When `disk_offload_dir` is `None`, uses the existing v2 path: - /// load per-shard RRDSHARD snapshot, replay per-shard WAL v2. + /// When `disk_offload_dir` is `None`, uses the legacy path: load the + /// per-shard RRDSHARD snapshot, then replay appendonly.aof (the recovery + /// AUTHORITY for this mode — WAL v3's KV coverage is intentionally + /// partial post-#211), with WAL v3 as a last-resort fallback only when + /// no AOF exists (the WAL v2 rung was removed in the pre-1.0 WAL-v3-only + /// format freeze — see `restore_from_persistence_v2`). /// - /// Returns total keys loaded (snapshot + WAL replay). + /// Returns total keys loaded (snapshot + AOF/WAL v3 replay). pub fn restore_from_persistence( &mut self, persistence_dir: &str, @@ -200,10 +204,27 @@ impl Shard { self.restore_from_persistence_v2(persistence_dir) } - /// V2 recovery path: snapshot load + WAL v2 replay + vector recovery. + /// Legacy recovery path: snapshot load + appendonly.aof (authority) / + /// WAL v3 (last-resort, AOF-absent-only) replay. + /// + /// Pre-1.0 WAL-v3-only format freeze: the per-shard WAL v2 rung + /// (`shard-N.wal`) was removed and is no longer replayed by this build — + /// see the loud `tracing::error!` below if one is still on disk. + /// + /// ⚠ The AOF — NOT WAL v3 — is the recovery authority here, and the old + /// WAL v2 "prefer the WAL over the AOF" contract must NOT be resurrected + /// with WAL v3: since PR #211 (`--wal-kv-log`, default auto=off when the + /// AOF is the authority) WAL v3 intentionally contains NO KV command + /// records in the default config — it carries CDC/PITR/temporal/graph + /// records — and even with `--wal-kv-log on`, connection-local writes + /// bypass it (measured WAL-only recovery: 79.2% incomplete, see + /// tmp/WRITE-DIAG.md). Preferring a non-empty WAL v3 and skipping the + /// AOF would therefore discard the only complete KV history. WAL v3 is + /// replayed ONLY when no `appendonly.aof` exists at all (disaster + /// fallback: partial recovery beats none), with a loud warning about + /// its partial KV coverage. fn restore_from_persistence_v2(&mut self, persistence_dir: &str) -> usize { use crate::persistence::snapshot::shard_snapshot_load; - use crate::persistence::wal; let dir = std::path::Path::new(persistence_dir); let mut total_keys = 0; @@ -222,43 +243,65 @@ impl Shard { } } - // Replay per-shard WAL, then fall back to appendonly.aof if WAL has 0 commands. - // The per-shard WalWriter writes to shard-N.wal but the global AOF writer - // (aof_writer_task) writes to appendonly.aof. Both may exist; try both. - let wal_file = wal::wal_path(dir, self.id); - let mut wal_replayed = 0usize; - if wal_file.exists() { - match wal::replay_wal(&mut self.databases, &wal_file, &DispatchReplayEngine::new()) { + // Loud failure (not silent skip) for a leftover pre-freeze WAL v2 + // file. WAL v2 support was removed in this build; the file below is + // NEVER consulted by any recovery path in this binary, so an + // operator who upgraded over a v2-only deployment must find out + // immediately rather than discover a silent recovery gap later. + let legacy_v2_wal = dir.join(format!("shard-{}.wal", self.id)); + if legacy_v2_wal.exists() { + tracing::error!( + "Shard {}: found legacy WAL v2 file {:?} — WAL v2 support was \ + removed (pre-1.0 WAL-v3-only format freeze) and this build \ + does NOT replay it. If its contents are not already reflected \ + in appendonly.aof, replay it on a pre-freeze Moon build first \ + (which re-persists through the AOF) before removing this file.", + self.id, + legacy_v2_wal + ); + } + + // AOF is the recovery authority (see doc comment: WAL v3 KV coverage + // is intentionally partial post-#211, so it must never shadow the AOF). + let aof_path = dir.join("appendonly.aof"); + if aof_path.exists() { + match crate::persistence::aof::replay_aof( + &mut self.databases, + &aof_path, + &DispatchReplayEngine::new(), + ) { Ok(n) => { - info!("Shard {}: replayed {} WAL commands", self.id, n); - wal_replayed = n; + info!("Shard {}: replayed {} AOF commands", self.id, n); total_keys += n; } Err(e) => { - tracing::error!("Shard {}: WAL replay failed: {}", self.id, e); + tracing::error!("Shard {}: AOF replay failed: {}", self.id, e); } } - } - // Fall back to appendonly.aof when per-shard WAL has 0 commands - if wal_replayed == 0 { - let aof_path = dir.join("appendonly.aof"); - if aof_path.exists() { - info!( - "Shard {}: WAL empty, falling back to appendonly.aof", - self.id - ); - match crate::persistence::aof::replay_aof( - &mut self.databases, - &aof_path, - &DispatchReplayEngine::new(), - ) { - Ok(n) => { - info!("Shard {}: replayed {} AOF commands", self.id, n); - total_keys += n; - } - Err(e) => { - tracing::error!("Shard {}: AOF replay failed: {}", self.id, e); - } + } else { + // Disaster fallback ONLY: no AOF at all (lost/rebuilt/never + // written). Partial recovery from WAL v3 beats nothing, but its + // KV coverage is incomplete by design — say so loudly. + let wal_v3_dir = dir.join(format!("shard-{}", self.id)).join("wal-v3"); + match crate::persistence::wal_v3::replay::replay_wal_v3_dir_commands( + &wal_v3_dir, + &mut self.databases, + &DispatchReplayEngine::new(), + ) { + Ok(0) => {} + Ok(n) => { + tracing::warn!( + "Shard {}: no appendonly.aof found — replayed {} WAL v3 \ + records as a LAST-RESORT fallback. WAL v3 KV coverage is \ + partial (gated by --wal-kv-log; connection-local writes \ + bypass it), so this recovery may be incomplete.", + self.id, + n + ); + total_keys += n; + } + Err(e) => { + tracing::error!("Shard {}: WAL v3 fallback replay failed: {}", self.id, e); } } } @@ -279,7 +322,6 @@ mod tests { use super::*; use crate::blocking::BlockingRegistry; use crate::framevec; - use crate::persistence::wal::WalWriter; use crate::protocol::Frame; use crate::pubsub::subscriber::Subscriber; use crate::runtime::channel as rt_channel; @@ -343,7 +385,6 @@ mod tests { let mut pending_snap = None; let mut snap_state = None; - let mut wal_w: Option = None; let blocking = Rc::new(RefCell::new(BlockingRegistry::new(0))); let script_cache = Rc::new(RefCell::new(crate::scripting::ScriptCache::new())); let clock = CachedClock::new(); @@ -355,8 +396,7 @@ mod tests { &blocking, &mut pending_snap, &mut snap_state, - &mut wal_w, - &mut None, // wal_v3_writer + &mut None, // wal_writer &backlog, &mut Vec::new(), &None, @@ -410,7 +450,6 @@ mod tests { let mut pending_snap = None; let mut snap_state = None; - let mut wal_w: Option = None; let blocking = Rc::new(RefCell::new(BlockingRegistry::new(0))); let script_cache = Rc::new(RefCell::new(crate::scripting::ScriptCache::new())); let clock = CachedClock::new(); @@ -422,8 +461,7 @@ mod tests { &blocking, &mut pending_snap, &mut snap_state, - &mut wal_w, - &mut None, // wal_v3_writer + &mut None, // wal_writer &backlog, &mut Vec::new(), &None, @@ -552,4 +590,140 @@ mod tests { &clock, ); } + + // ── restore_from_persistence_v2: AOF is the recovery authority; WAL v3 + // is a last-resort fallback ONLY when no AOF exists (review finding P0#1, + // corrected) ──────────────────────────────────────────────────────────── + // + // WAL v3 is written even without --disk-offload, but post-#211 its KV + // coverage is intentionally PARTIAL (`--wal-kv-log` default-off; + // connection-local writes bypass it), so it must never shadow the AOF. + // See `src/persistence/recovery.rs`'s equivalent tests for the + // disk-offload path's analogous fallback. + + #[test] + fn test_restore_from_persistence_v2_aof_is_authority_over_wal_v3() { + use crate::persistence::wal_v3::record::{WalRecordType, write_wal_v3_record}; + + let tmp = tempfile::tempdir().unwrap(); + let dir = tmp.path(); + + // Populated `shard-0/wal-v3/` (3 records). Post-#211 the WAL may hold + // MORE records than the AOF while still missing KV writes entirely — + // record count says nothing about KV completeness. + let wal_dir = dir.join("shard-0").join("wal-v3"); + std::fs::create_dir_all(&wal_dir).unwrap(); + let mut header = vec![0u8; 64]; + header[0..6].copy_from_slice(b"RRDWAL"); + header[6] = 3; // version = 3 + let mut wal_data = header; + for i in 1..=3u64 { + write_wal_v3_record( + &mut wal_data, + i, + WalRecordType::Command, + b"*1\r\n$4\r\nPING\r\n", + ); + } + std::fs::write(wal_dir.join("000000000001.wal"), &wal_data).unwrap(); + + // The 1-record appendonly.aof is the authority and MUST win. + std::fs::write(dir.join("appendonly.aof"), b"*1\r\n$4\r\nPING\r\n").unwrap(); + + let config = RuntimeConfig::default(); + let mut shard = Shard::new(0, 1, 1, config); + let total = shard.restore_from_persistence_v2(dir.to_str().unwrap()); + + assert_eq!( + total, 1, + "restore_from_persistence_v2 must replay the authoritative AOF \ + and ignore WAL v3 when the AOF exists -- got {} (a non-empty \ + but KV-incomplete WAL v3 shadowed the AOF if this is 3)", + total + ); + } + + #[test] + fn test_restore_from_persistence_v2_wal_v3_last_resort_when_no_aof() { + use crate::persistence::wal_v3::record::{WalRecordType, write_wal_v3_record}; + + let tmp = tempfile::tempdir().unwrap(); + let dir = tmp.path(); + + // WAL v3 present, NO appendonly.aof at all: partial recovery from + // the WAL beats recovering nothing. + let wal_dir = dir.join("shard-0").join("wal-v3"); + std::fs::create_dir_all(&wal_dir).unwrap(); + let mut header = vec![0u8; 64]; + header[0..6].copy_from_slice(b"RRDWAL"); + header[6] = 3; // version = 3 + let mut wal_data = header; + for i in 1..=3u64 { + write_wal_v3_record( + &mut wal_data, + i, + WalRecordType::Command, + b"*1\r\n$4\r\nPING\r\n", + ); + } + std::fs::write(wal_dir.join("000000000001.wal"), &wal_data).unwrap(); + + let config = RuntimeConfig::default(); + let mut shard = Shard::new(0, 1, 1, config); + let total = shard.restore_from_persistence_v2(dir.to_str().unwrap()); + + assert_eq!( + total, 3, + "with no appendonly.aof, the WAL v3 last-resort fallback must \ + replay what it can" + ); + } + + #[test] + fn test_restore_from_persistence_v2_falls_back_to_aof_when_wal_v3_absent() { + let tmp = tempfile::tempdir().unwrap(); + let dir = tmp.path(); + + // No `shard-0/wal-v3/` at all -- only the AOF. + std::fs::write( + dir.join("appendonly.aof"), + b"*1\r\n$4\r\nPING\r\n*1\r\n$4\r\nPING\r\n", + ) + .unwrap(); + + let config = RuntimeConfig::default(); + let mut shard = Shard::new(0, 1, 1, config); + let total = shard.restore_from_persistence_v2(dir.to_str().unwrap()); + + assert_eq!( + total, 2, + "with no WAL v3 data at all, the AOF fallback must still work \ + exactly as before this fix" + ); + } + + #[test] + fn test_restore_from_persistence_v2_stray_legacy_v2_wal_does_not_break_aof() { + let tmp = tempfile::tempdir().unwrap(); + let dir = tmp.path(); + + // Stray pre-freeze WAL v2 file (`shard-0.wal`, RRDWAL version=2) must + // be loudly logged and ignored, not crash recovery or block the AOF + // fallback. + let mut legacy_v2_wal = vec![0u8; 32]; + legacy_v2_wal[0..6].copy_from_slice(b"RRDWAL"); + legacy_v2_wal[6] = 2; + std::fs::write(dir.join("shard-0.wal"), &legacy_v2_wal).unwrap(); + std::fs::write(dir.join("appendonly.aof"), b"*1\r\n$4\r\nPING\r\n").unwrap(); + + let config = RuntimeConfig::default(); + let mut shard = Shard::new(0, 1, 1, config); + let total = shard.restore_from_persistence_v2(dir.to_str().unwrap()); + + assert_eq!( + total, 1, + "a stray legacy v2 WAL file must not prevent the AOF fallback \ + from working" + ); + } } diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index 4c93ae3a..7f1bf5a4 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -8,7 +8,6 @@ use std::sync::Arc; use tracing::info; use crate::persistence::snapshot::SnapshotState; -use crate::persistence::wal::WalWriter; use crate::runtime::channel; use super::shared_databases::ShardDatabases; @@ -147,33 +146,21 @@ pub(crate) fn advance_snapshot_segment( } } -/// Handle successful snapshot finalization: truncate WAL and send reply. +/// Handle successful snapshot finalization: send reply. +/// +/// WAL v2's per-snapshot `truncate_after_snapshot(epoch)` had no v3 +/// equivalent -- v3 retention is LSN-driven (`WalWriterV3::recycle_aggressive` +/// / `recycle_segments_before`, invoked from autovacuum Pass C and the +/// checkpoint protocol) and runs independently of legacy RRDSHARD snapshot +/// epochs, so this handler no longer touches the WAL writer at all. pub(crate) fn finalize_snapshot_success( snapshot_state: &mut Option, snapshot_reply_tx: &mut Option>>, - wal_writer: &mut Option, shard_id: usize, ) { if let Some(snap) = snapshot_state.as_ref() { let epoch = snap.epoch; info!("Shard {}: snapshot epoch {} complete", shard_id, epoch); - // Truncate WAL after successful snapshot - if let Some(wal) = wal_writer { - if let Err(e) = wal.truncate_after_snapshot(epoch) { - tracing::error!( - "Shard {}: WAL truncation after snapshot epoch {} failed: {}. \ - WAL and snapshot may be out of sync.", - shard_id, - epoch, - e - ); - if let Some(tx) = snapshot_reply_tx.take() { - let _ = tx.send(Err(format!("WAL truncation failed: {}", e))); - } - *snapshot_state = None; - return; - } - } if let Some(tx) = snapshot_reply_tx.take() { let _ = tx.send(Ok(())); } @@ -195,18 +182,11 @@ pub(crate) fn finalize_snapshot_error( *snapshot_state = None; } -/// Flush WAL if needed (1ms tick -- write to page cache only; sync is separate). -pub(crate) fn flush_wal_if_needed(wal_writer: &mut Option) { - if let Some(wal) = wal_writer { - if let Err(e) = wal.flush_if_needed() { - tracing::error!("WAL flush failed: {}", e); - } - } -} - -/// Flush WAL v3 if buffer exceeds threshold (1ms tick -- mirrors v2 pattern). +/// Flush WAL if buffer exceeds threshold (1ms tick -- write to page cache +/// only; durable sync is separate, see `timers::sync_wal_v3`). /// -/// Only active when disk-offload is enabled and WalWriterV3 was successfully initialized. +/// Only active when the per-shard WAL writer was successfully initialized +/// (appendonly=yes; see the writer-creation block in `event_loop::run`). pub(crate) fn flush_wal_v3_if_needed( wal_v3: &mut Option, ) { diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs index 8797d81c..cc455e9f 100644 --- a/src/shard/shared_databases.rs +++ b/src/shard/shared_databases.rs @@ -45,7 +45,7 @@ pub struct ShardStoreMemory { /// - `store_memory_per_shard`: published store-memory atomics (C5 / M4). pub struct ShardDatabases { /// Per-shard WAL append channel sender. Connection handlers send serialized - /// write commands here; the event loop drains into WAL v2/v3 on the 1ms tick. + /// write commands here; the event loop drains into WAL v3 on the 1ms tick. /// OnceLock: set once at event-loop startup (before connections are /// accepted), then every hot-path read is lock-free. wal_append_txs: Vec>>, @@ -154,7 +154,7 @@ impl ShardDatabases { /// Send serialized command bytes to the WAL append channel for a shard. /// /// Called by connection handlers for local write commands. The event loop - /// drains this channel on the 1ms tick into WAL v2/v3. + /// drains this channel on the 1ms tick into WAL v3. /// No-op when persistence is disabled. #[inline] pub fn wal_append(&self, shard_id: usize, data: bytes::Bytes) { @@ -544,40 +544,73 @@ pub fn recover_graph_stores( } /// Replay graph WAL commands into graph stores for all shards. +/// +/// Pre-1.0 WAL-v3-only format freeze: ported from the deleted WAL v2 flat +/// file (`shard-{id}.wal`) to the v3 segment directory +/// (`shard-{id}/wal-v3/*.wal`, matching `replay_workspace_wal` / +/// `replay_temporal_wal` above and `recovery.rs`). `Command` records are +/// forwarded to `DispatchReplayEngine::replay_command` exactly as +/// `replay_wal_auto`'s v3 branch does, so this stays bug-for-bug consistent +/// with the rest of the WAL v3 command-replay path. #[cfg(feature = "graph")] pub fn replay_graph_wal( inits: &mut [crate::shard::slice::ShardSliceInit], persistence_dir: &std::path::Path, db_count: usize, ) { - use crate::persistence::replay::DispatchReplayEngine; - use crate::persistence::wal; + use crate::persistence::replay::{CommandReplayEngine, DispatchReplayEngine}; + use crate::persistence::wal_v3::record::{WalRecord, WalRecordType}; for init in inits.iter_mut() { let shard_id = init.shard_id; - let wal_file = wal::wal_path(persistence_dir, shard_id); - if !wal_file.exists() { + let wal_dir = persistence_dir + .join(format!("shard-{}", shard_id)) + .join("wal-v3"); + if !wal_dir.exists() { continue; } + let engine = DispatchReplayEngine::new(); let mut dummy_dbs: Vec = (0..db_count).map(|_| Database::new()).collect(); - match wal::replay_wal(&mut dummy_dbs, &wal_file, &engine) { - Ok(_) => { - let graph_count = engine.graph_command_count(); - if graph_count > 0 { - let applied = engine.replay_graph_commands(&mut init.graph_store); - tracing::info!( - "Shard {}: replayed {} graph WAL commands ({} applied)", - shard_id, - graph_count, - applied, - ); - } + let mut selected_db = 0usize; + let on_command = &mut |record: &WalRecord| { + if record.record_type == WalRecordType::Command { + engine.replay_command(&mut dummy_dbs, &record.payload, &[], &mut selected_db); } + }; + let on_fpi = &mut |_record: &WalRecord| {}; + + let mut wal_files: Vec<_> = match std::fs::read_dir(&wal_dir) { + Ok(entries) => entries + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_str().is_some_and(|n| n.ends_with(".wal"))) + .map(|e| e.path()) + .collect(), Err(e) => { + tracing::error!("Shard {}: graph WAL dir read failed: {}", shard_id, e); + continue; + } + }; + wal_files.sort(); + + for wal_file in &wal_files { + if let Err(e) = crate::persistence::wal_v3::replay::replay_wal_v3_file( + wal_file, 0, on_command, on_fpi, + ) { tracing::error!("Shard {}: graph WAL replay failed: {}", shard_id, e); } } + + let graph_count = engine.graph_command_count(); + if graph_count > 0 { + let applied = engine.replay_graph_commands(&mut init.graph_store); + tracing::info!( + "Shard {}: replayed {} graph WAL commands ({} applied)", + shard_id, + graph_count, + applied, + ); + } } } diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 0de13410..e5fa32a9 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -17,7 +17,6 @@ use crate::command::{DispatchResult, dispatch as cmd_dispatch}; use crate::config::RuntimeConfig; use crate::persistence::aof; use crate::persistence::snapshot::SnapshotState; -use crate::persistence::wal::WalWriter; use crate::persistence::wal_v3::segment::WalWriterV3; use crate::pubsub::PubSubRegistry; use crate::replication::backlog::ReplicationBacklog; @@ -93,8 +92,7 @@ pub(crate) fn drain_spsc_shared( channel::OneshotSender>, )>, snapshot_state: &mut Option, - wal_writer: &mut Option, - wal_v3_writer: &mut Option, + wal_writer: &mut Option, repl_backlog: &crate::replication::backlog::SharedBacklog, replica_txs: &mut Vec<(u64, channel::MpscSender)>, repl_state: &Option, @@ -220,7 +218,7 @@ pub(crate) fn drain_spsc_shared( // touch the registry through handle_shard_message // because the registry's WalTailReader needs the // shard's wal_dir, which the event loop already - // has from wal_v3_writer.wal_dir(). + // has from wal_writer.wal_dir(). pending_cdc_subscribes.push(*payload); } _ => other_messages.push(msg), @@ -245,7 +243,6 @@ pub(crate) fn drain_spsc_shared( pending_snapshot, snapshot_state, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -282,7 +279,6 @@ pub(crate) fn drain_spsc_shared( pending_snapshot, snapshot_state, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -332,8 +328,7 @@ pub(crate) fn handle_shard_message_shared( channel::OneshotSender>, )>, snapshot_state: &mut Option, - wal_writer: &mut Option, - wal_v3_writer: &mut Option, + wal_writer: &mut Option, repl_backlog: &crate::replication::backlog::SharedBacklog, replica_txs: &mut Vec<(u64, channel::MpscSender)>, repl_state: &Option, @@ -498,7 +493,7 @@ pub(crate) fn handle_shard_message_shared( crate::command::server_admin::vacuum( &mut s.vector_store, shard_manifest.as_mut(), - wal_v3_writer.as_mut(), + wal_writer.as_mut(), args, mvcc_prune_margin, ) @@ -517,7 +512,7 @@ pub(crate) fn handle_shard_message_shared( crate::command::server_admin::debug_reclamation( &s.vector_store, shard_manifest.as_ref(), - wal_v3_writer.as_ref(), + wal_writer.as_ref(), ) }); let _ = reply_tx.send(frame); @@ -578,7 +573,6 @@ pub(crate) fn handle_shard_message_shared( if !wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -655,7 +649,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -809,7 +802,6 @@ pub(crate) fn handle_shard_message_shared( if matches!(response, crate::protocol::Frame::Integer(1)) && wal_fanout_has_work( wal_writer, - wal_v3_writer, replica_txs, aof_pool, wal_kv_log, @@ -819,7 +811,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -877,18 +868,11 @@ pub(crate) fn handle_shard_message_shared( // Skip the serialization alloc when the fanout would // no-op (persistence + replication all off) — it was // pure waste on every cross-shard write. - if wal_fanout_has_work( - wal_writer, - wal_v3_writer, - replica_txs, - aof_pool, - wal_kv_log, - ) { + if wal_fanout_has_work(wal_writer, replica_txs, aof_pool, wal_kv_log) { let serialized = aof::serialize_command(cmd_frame); aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1001,7 +985,6 @@ pub(crate) fn handle_shard_message_shared( if matches!(response, crate::protocol::Frame::Integer(1)) && wal_fanout_has_work( wal_writer, - wal_v3_writer, replica_txs, aof_pool, wal_kv_log, @@ -1011,7 +994,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1068,18 +1050,11 @@ pub(crate) fn handle_shard_message_shared( if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) { // See `wal_fanout_has_work` — skip the serialization alloc // entirely when the fanout would no-op (persistence off). - if wal_fanout_has_work( - wal_writer, - wal_v3_writer, - replica_txs, - aof_pool, - wal_kv_log, - ) { + if wal_fanout_has_work(wal_writer, replica_txs, aof_pool, wal_kv_log) { let serialized = aof::serialize_command(cmd_frame); aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1236,7 +1211,6 @@ pub(crate) fn handle_shard_message_shared( if !wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1305,7 +1279,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1420,7 +1393,6 @@ pub(crate) fn handle_shard_message_shared( if matches!(response, crate::protocol::Frame::Integer(1)) && wal_fanout_has_work( wal_writer, - wal_v3_writer, replica_txs, aof_pool, wal_kv_log, @@ -1430,7 +1402,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1488,18 +1459,11 @@ pub(crate) fn handle_shard_message_shared( // Skip the serialization alloc when the fanout would // no-op (persistence + replication all off) — it was // pure waste on every cross-shard write. - if wal_fanout_has_work( - wal_writer, - wal_v3_writer, - replica_txs, - aof_pool, - wal_kv_log, - ) { + if wal_fanout_has_work(wal_writer, replica_txs, aof_pool, wal_kv_log) { let serialized = aof::serialize_command(cmd_frame); aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1613,7 +1577,6 @@ pub(crate) fn handle_shard_message_shared( if matches!(response, crate::protocol::Frame::Integer(1)) && wal_fanout_has_work( wal_writer, - wal_v3_writer, replica_txs, aof_pool, wal_kv_log, @@ -1623,7 +1586,6 @@ pub(crate) fn handle_shard_message_shared( aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -1680,18 +1642,11 @@ pub(crate) fn handle_shard_message_shared( if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) { // See `wal_fanout_has_work` — skip the serialization alloc // entirely when the fanout would no-op (persistence off). - if wal_fanout_has_work( - wal_writer, - wal_v3_writer, - replica_txs, - aof_pool, - wal_kv_log, - ) { + if wal_fanout_has_work(wal_writer, replica_txs, aof_pool, wal_kv_log) { let serialized = aof::serialize_command(cmd_frame); aof_ok = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -2241,7 +2196,7 @@ pub(crate) fn handle_shard_message_shared( ShardMessage::SwapDb { a, b, reply_tx } => { // WAL-before-swap: emit the SWAPDB record so that crash-recovery // replay can re-apply the swap in the correct order. The record - // is written even when wal_writer/wal_v3_writer are None (the + // is written even when wal_writer/wal_writer are None (the // fast-path in wal_append_and_fanout will skip it cheaply). // // Serialise "SWAPDB " without heap allocation on the number @@ -2263,7 +2218,6 @@ pub(crate) fn handle_shard_message_shared( let _ = wal_append_and_fanout( &serialized, wal_writer, - wal_v3_writer, repl_backlog, replica_txs, repl_state, @@ -3124,15 +3078,15 @@ pub(crate) fn cow_intercept( /// fan-out to all connected replica sender channels (non-blocking try_send), and route /// the entry to the per-shard AOF writer pool when AOF is enabled. /// -/// CRITICAL: shard_offset in ReplicationState is SEPARATE from WalWriter::bytes_written. -/// WalWriter::bytes_written resets on snapshot truncation; shard_offset NEVER resets. +/// CRITICAL: shard_offset in ReplicationState is SEPARATE from WalWriterV3::bytes_written. +/// WalWriterV3's on-disk bytes reset only on segment recycling; shard_offset NEVER resets. /// /// FIX-W1-2: `aof_pool` was added to route MSET/coordinator cross-shard writes /// through the per-shard AOF pool. The SPSC drain is synchronous so we use /// `try_send_append` (fire-and-forget). The `appendfsync=always` rendezvous is /// handled by the connection handler (async context), not here. /// True when `wal_append_and_fanout` has any consumer for the serialized -/// command (S3.5b criterion: WAL v2/v3, live replicas, or the AOF pool). +/// command (S3.5b criterion: WAL, live replicas, or the AOF pool). /// ARM perf annotate showed the pre-S3.5b locks were ~21% of CPU on 8-shard /// SET p=64 with everything off; the criterion is fully derivable from the /// inputs — no flags or shared state. Skipping leaves shard_offset @@ -3143,15 +3097,12 @@ pub(crate) fn cow_intercept( /// every cross-shard write with persistence off). #[inline] pub(crate) fn wal_fanout_has_work( - wal_writer: &Option, - wal_v3_writer: &Option, + wal_writer: &Option, replica_txs: &[(u64, channel::MpscSender)], aof_pool: Option<&std::sync::Arc>, wal_kv_log: bool, ) -> bool { - (wal_kv_log && (wal_writer.is_some() || wal_v3_writer.is_some())) - || !replica_txs.is_empty() - || aof_pool.is_some() + (wal_kv_log && wal_writer.is_some()) || !replica_txs.is_empty() || aof_pool.is_some() } /// Error frame substituted for a write's success frame when the command @@ -3170,8 +3121,7 @@ pub(crate) const AOF_APPEND_LOST_ERR: &[u8] = /// per drain arm, not per command. pub(crate) fn wal_append_and_fanout( data: &[u8], - wal_writer: &mut Option, - wal_v3_writer: &mut Option, + wal_writer: &mut Option, repl_backlog: &crate::replication::backlog::SharedBacklog, replica_txs: &[(u64, channel::MpscSender)], repl_state: &Option, @@ -3183,7 +3133,7 @@ pub(crate) fn wal_append_and_fanout( // S3.5b (2026-04-27): hot-path bypass when nothing actually has work. // See `wal_fanout_has_work` — callers use the same predicate to skip the // `aof::serialize_command` alloc entirely when the fanout would no-op. - if !wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool, wal_kv_log) { + if !wal_fanout_has_work(wal_writer, replica_txs, aof_pool, wal_kv_log) { return true; } // `wal_kv_log == false` (--wal-kv-log auto/off): the AOF is the recovery @@ -3192,15 +3142,11 @@ pub(crate) fn wal_append_and_fanout( // pure write amplification (measured 2.7× file bytes at shards=4). // FPI/checkpoint/feature records are unaffected (different entry points). if wal_kv_log { - // WAL v3 supersedes v2 — skip v2 append when v3 is active to avoid - // double-write overhead (2 write syscalls per SPSC drain batch). - if let Some(w3) = wal_v3_writer { - w3.append( + if let Some(w) = wal_writer { + w.append( crate::persistence::wal_v3::record::WalRecordType::Command, data, ); - } else if let Some(w) = wal_writer { - w.append(data); } } // 2. Replication backlog (in-memory circular buffer for partial resync). @@ -3285,8 +3231,7 @@ mod wal_append_tests { wal_append_and_fanout( b"hello", - &mut None, // no v2 writer - &mut None, // no v3 writer + &mut None, // no writer &backlog, &[], // no replicas &None, // no repl_state @@ -3316,7 +3261,6 @@ mod wal_append_tests { wal_append_and_fanout( b"hello", &mut None, - &mut None, &backlog, &replica_txs, &None, @@ -3354,8 +3298,7 @@ mod wal_append_tests { wal_append_and_fanout( b"world", - &mut None, // no v2 writer - &mut None, // no v3 writer + &mut None, // no writer &backlog, &[], // no replicas — S3.5b bypass triggered without pool guard &None, // no repl_state @@ -3419,8 +3362,7 @@ mod wal_append_tests { // Pre-fix this was `aof_pool` (Some), which caused the double-write. wal_append_and_fanout( b"*3\r\n$3\r\nSET\r\n$1\r\na\r\n$1\r\n1\r\n", - &mut None, // no v2 writer - &mut None, // no v3 writer + &mut None, // no writer &backlog, &[], // no replicas &None, // no repl_state @@ -3446,7 +3388,6 @@ mod wal_append_tests { wal_append_and_fanout( b"*3\r\n$4\r\nMSET\r\n$1\r\nb\r\n$1\r\n2\r\n", &mut None, - &mut None, &backlog, &[], &None, @@ -3493,7 +3434,6 @@ mod wal_append_tests { wal_append_and_fanout( cmd, - &mut None, &mut w3, &backlog, &[], @@ -3518,7 +3458,6 @@ mod wal_append_tests { // the record must land in the WAL as before. wal_append_and_fanout( cmd, - &mut None, &mut w3, &backlog, &[], @@ -3569,8 +3508,7 @@ mod drain_cap_tests { let blocking = Rc::new(RefCell::new(BlockingRegistry::new(0))); let mut pending_snapshot = None; let mut snapshot_state: Option = None; - let mut wal_writer: Option = None; - let mut wal_v3_writer: Option = None; + let mut wal_writer: Option = None; let backlog: crate::replication::backlog::SharedBacklog = Arc::new(parking_lot::Mutex::new(None)); let mut replica_txs = Vec::new(); @@ -3596,7 +3534,6 @@ mod drain_cap_tests { &mut pending_snapshot, &mut snapshot_state, &mut wal_writer, - &mut wal_v3_writer, &backlog, &mut replica_txs, &offsets, @@ -3631,7 +3568,6 @@ mod drain_cap_tests { &mut pending_snapshot, &mut snapshot_state, &mut wal_writer, - &mut wal_v3_writer, &backlog, &mut replica_txs, &offsets, diff --git a/src/shard/timers.rs b/src/shard/timers.rs index 50498bf4..786d90ed 100644 --- a/src/shard/timers.rs +++ b/src/shard/timers.rs @@ -9,7 +9,6 @@ use std::sync::Arc; use crate::blocking::BlockingRegistry; use crate::config::RuntimeConfig; -use crate::persistence::wal::WalWriter; use super::shared_databases::ShardDatabases; @@ -71,15 +70,6 @@ pub const CHECKPOINT_TICK_MS: u64 = 1; /// Infrequent enough to avoid overhead, responsive enough to catch aged segments. pub const WARM_CHECK_INTERVAL_MS: u64 = 10_000; -/// WAL fsync on 1-second interval (everysec durability). -pub(crate) fn sync_wal(wal_writer: &mut Option) { - if let Some(wal) = wal_writer { - if let Err(e) = wal.sync_to_disk() { - tracing::error!("WAL sync failed: {}", e); - } - } -} - /// Fire MQ triggers whose debounce window has elapsed. /// /// Called from the event loop periodic tick at 100ms cadence (same interval diff --git a/tests/crash_matrix_per_shard_aof.rs b/tests/crash_matrix_per_shard_aof.rs index fcffbc0c..5e5127c3 100644 --- a/tests/crash_matrix_per_shard_aof.rs +++ b/tests/crash_matrix_per_shard_aof.rs @@ -459,3 +459,43 @@ fn pipeline_batch_no_double_write_after_crash_recovery() { let _ = std::fs::remove_dir_all(&dir); } + +// NOTE on NONOFFLOAD-WAL-V3-FALLBACK (dropped, see recovery.rs / +// wal_v3::replay unit tests instead): +// +// An earlier version of this file had a SIGKILL-based integration test here +// for the `restore_from_persistence_v2` / `recover_shard_v3_pitr` legacy-dir +// recovery fix (see `src/persistence/recovery.rs` and `src/shard/mod.rs`). +// It was removed after empirical testing (manual server runs against this +// exact binary) showed it could not reliably exercise the fix: +// +// 1. `--disk-offload` defaults to `enable`, so a plain `--shards 1 +// --appendonly yes` server takes the DISK-OFFLOAD recovery path +// (`recover_shard_v3_with_fallback`), not `restore_from_persistence_v2` -- +// the flag must be passed explicitly. +// 2. Ordinary local (same-shard-as-connection) KV commands (SET/DEL/...) are +// durability-logged ONLY to the multi-part AOF +// (`aof_pool.try_send_append_durable`, see `handler_monoio/mod.rs`) -- +// they never reach WAL v3's `Command` record path at all. WAL v3 only +// receives KV data for CROSS-SHARD SPSC-dispatched writes (gated by +// `--wal-kv-log`) or WS/MQ/GRAPH/cross-store-TXN records. With `--shards +// 1` there is no cross-shard dispatch, so WAL v3 stays a bare 64-byte +// header regardless of `--wal-kv-log`. +// 3. Forcing cross-shard dispatch with `--shards 2` is viable (verified: a +// shard's `wal-v3` segment DOES grow past the header when it receives +// cross-shard writes with `--wal-kv-log on`), but which shard receives +// "local" vs "cross-shard" traffic depends on which shard SO_REUSEPORT +// assigns each new connection to -- non-deterministic across separate +// `redis-cli` invocations, so an assertion over "every key recovered" +// cannot be made reliable without a custom persistent-connection client +// pinned to a known shard. +// +// The fix itself (legacy-dir recovery replaying the authoritative AOF and +// falling back to the legacy-mode WAL v3 dir ONLY when no AOF exists — +// WAL v3's KV coverage is partial post-#211, so it must never shadow the +// AOF) is instead covered by deterministic +// unit tests in `src/persistence/recovery.rs` and +// `src/persistence/wal_v3/replay.rs`, which construct WAL v3 segments +// directly (the same pattern every other recovery test in this codebase +// already uses) rather than depending on which redis-cli command happens to +// reach WAL v3 in a live server.