diff --git a/CHANGELOG.md b/CHANGELOG.md index a352a5b9d..d5a84d358 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **The AOF now compacts itself (#433): Redis-parity automatic rewrite.** + `--auto-aof-rewrite-percentage` (default 100, `0` disables) and + `--auto-aof-rewrite-min-size` (default `64mb`, size strings accepted) + trigger a background rewrite once the AOF has grown the given percentage + over its size after the last rewrite. Before this, the AOF grew with write + volume rather than dataset size — observed 4.8 GB on disk for a 2.43 GB + dataset, ~1 GB/day — until the diskfull guard paused writes. A monitor + thread samples the on-disk size once a second and dispatches the same + entry point as `BGREWRITEAOF`; a failed dispatch backs off 60 s instead of + hot-retrying. Both knobs appear in `CONFIG GET`. +- **Multi-shard `BGREWRITEAOF` is un-gated.** The per-shard fan-out rewrite + (cooperative snapshot + synchronized manifest commit) is now the default — + the historical gate dated from a pre-C4 design that lost ~38% of keys, and + the current path holds exact INCR recovery across a rewrite straddling a + live write stream plus SIGKILL (crash matrix, 5/5 repeat runs). + `--experimental-per-shard-rewrite` is deprecated (warns, no-op). + ### Fixed +- **`INFO persistence` reports real AOF state (#432).** `aof_enabled` and + `aof_rewrite_in_progress` were hardcoded `0` even with `--appendonly yes` + (the default) and a rewrite running; they now reflect reality, and new + `aof_base_size` / `aof_current_size` fields expose the growth the + auto-rewrite trigger acts on — an operator can finally see the + AOF-vs-dataset ratio the diskfull incident hid. +- **The per-shard BGREWRITEAOF crash matrix no longer reports phantom data + loss on nearly-full hosts.** The harness lacked `--disk-free-min-pct 0` + and parsed `MOONERR diskfull` INCR rejections as silently-dropped writes + (the host root volume hovers at ~4% free, making it intermittent). The + suite now disables the guard, panics on any non-numeric INCR reply, and is + green 5/5 consecutive runs. - **Replicas now apply streamed `SWAPDB` (#386), and the record reaches the wire exactly once per client call.** Two stacked defects: (1) the replica's apply path had no SWAPDB intercept — generic dispatch hard-errors ("must be diff --git a/docs/runbooks/multi-shard-aof-rewrite.md b/docs/runbooks/multi-shard-aof-rewrite.md index b751ab005..499618396 100644 --- a/docs/runbooks/multi-shard-aof-rewrite.md +++ b/docs/runbooks/multi-shard-aof-rewrite.md @@ -52,15 +52,25 @@ linearly with shard count. ### BGREWRITEAOF in per-shard mode -`BGREWRITEAOF` is **not yet supported** for PerShard layouts. Issuing it on a -PerShard instance returns the following error immediately: - -``` -ERR BGREWRITEAOF is not yet supported under per-shard AOF layout; per-shard rewrite ships in step 6 of the per-shard AOF migration -``` - -Per-shard BGREWRITEAOF (each shard compacts its own log independently, with -all N acks awaited before returning confirmation) is tracked for v0.2. +`BGREWRITEAOF` is **fully supported** on PerShard layouts (#433, un-gated +2026-08): the per-shard fan-out folds every shard cooperatively (C4 snapshot +via the shard event loop), advances each shard's manifest entry, and the last +writer performs a single synchronized seq commit + old-generation prune. +Exactness (no dropped, no double-applied writes across a rewrite that +straddles a live write stream + SIGKILL) is pinned by +`tests/crash_matrix_per_shard_bgrewriteaof.rs`. +`--experimental-per-shard-rewrite`, the old opt-in, is deprecated and a +no-op. + +### Automatic rewrite (#433) + +The AOF compacts itself, Redis-style: when its on-disk size exceeds +`auto-aof-rewrite-min-size` (default `64mb`) AND has grown +`auto-aof-rewrite-percentage`% (default `100`) over its size after the last +rewrite, a background rewrite is dispatched automatically. Set the +percentage to `0` to disable automatic rewrites (manual `BGREWRITEAOF` +always works). Watch `INFO persistence` → `aof_base_size` / +`aof_current_size` to see the trigger's inputs. --- diff --git a/src/command/config.rs b/src/command/config.rs index 9173958b9..05a840a37 100644 --- a/src/command/config.rs +++ b/src/command/config.rs @@ -41,6 +41,17 @@ pub fn config_get( ), (b"appendonly", runtime_config.appendonly.clone()), (b"appendfsync", runtime_config.appendfsync.clone()), + ( + b"auto-aof-rewrite-percentage", + server_config.auto_aof_rewrite_percentage.to_string(), + ), + ( + // Redis reports this in bytes; normalize the "64mb"-style input. + b"auto-aof-rewrite-min-size", + crate::config::ServerConfig::parse_size(&server_config.auto_aof_rewrite_min_size) + .unwrap_or(64 * 1024 * 1024) + .to_string(), + ), (b"databases", server_config.databases.to_string()), (b"bind", server_config.bind.clone()), (b"port", server_config.port.to_string()), diff --git a/src/command/connection.rs b/src/command/connection.rs index 88ae7c343..c716ec31c 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -278,13 +278,26 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { sections.push_str("\r\n"); sections.push_str("# Persistence\r\n"); + // #432: aof_enabled / aof_rewrite_in_progress / sizes are real state, not + // hardcoded zeros. Sizes come from the auto-rewrite monitor's statics + // (#433); refresh_current_size keeps `aof_current_size` honest when INFO + // is read between monitor ticks (one directory walk — INFO is cold path). + let aof_enabled = crate::persistence::aof::auto_rewrite::AOF_ENABLED + .load(std::sync::atomic::Ordering::Relaxed); + let aof_current_size = if aof_enabled { + crate::persistence::aof::auto_rewrite::refresh_current_size() + } else { + 0 + }; sections.push_str(&format!( "loading:0\r\n\ rdb_bgsave_in_progress:{}\r\n\ rdb_last_save_time:{}\r\n\ rdb_last_bgsave_status:{}\r\n\ - aof_enabled:0\r\n\ - aof_rewrite_in_progress:0\r\n\ + aof_enabled:{}\r\n\ + aof_rewrite_in_progress:{}\r\n\ + aof_base_size:{}\r\n\ + aof_current_size:{}\r\n\ aof_backpressure_dropped:{}\r\n\ spill_batches_flushed:{}\r\n\ spill_completions_dropped:{}\r\n\ @@ -303,6 +316,14 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame { } else { "err" }, + u8::from(aof_enabled), + u8::from( + crate::command::persistence::AOF_REWRITE_IN_PROGRESS + .load(std::sync::atomic::Ordering::SeqCst) + ), + crate::persistence::aof::auto_rewrite::AOF_BASE_SIZE + .load(std::sync::atomic::Ordering::Relaxed), + aof_current_size, crate::persistence::aof::AOF_BACKPRESSURE_DROPPED .load(std::sync::atomic::Ordering::Relaxed), crate::storage::tiered::spill_thread::spill_batches_flushed_total(), diff --git a/src/config.rs b/src/config.rs index aafe54db4..82ef270f0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -117,19 +117,11 @@ pub struct ServerConfig { #[arg(long, default_value_t = false)] pub unsafe_multishard_aof: bool, - /// [EXPERIMENTAL] Enable per-shard BGREWRITEAOF (compaction) for the - /// `--shards >= 2 + --appendonly yes` PerShard layout. - /// - /// Default `false`: BGREWRITEAOF stays gated in PerShard mode (the - /// shipped, crash-safe "append-only, no in-place compaction" behavior). - /// When `true`, BGREWRITEAOF fans the rewrite out to every per-shard - /// writer (synchronized seq bump + single manifest commit). This path is - /// validated by `tests/crash_matrix_per_shard_bgrewriteaof.rs` and is - /// opt-in until the both-runtime crash matrix is green by default. - /// - /// The flag only takes effect alongside `per_shard_aof_active`; it is a - /// no-op for `--shards 1` (TopLevel rewrite already works) and for - /// `--appendonly no`. + /// [DEPRECATED — no-op] Per-shard BGREWRITEAOF is the DEFAULT since #433 + /// (the fan-out compaction path is validated by + /// `tests/crash_matrix_per_shard_bgrewriteaof.rs` and the auto-rewrite + /// suite). Passing this flag only emits a deprecation warning. Remove it + /// from launch commands; it will be deleted in a future release. #[arg(long, default_value_t = false)] pub experimental_per_shard_rewrite: bool, @@ -137,6 +129,20 @@ pub struct ServerConfig { #[arg(long, default_value = "everysec")] pub appendfsync: String, + /// Automatic AOF rewrite trigger: rewrite when the AOF has grown by this + /// percentage over its size after the last rewrite (Redis parity: + /// `auto-aof-rewrite-percentage`). `0` disables automatic rewrites; + /// manual `BGREWRITEAOF` still works. Default 100 (= rewrite at 2× the + /// post-rewrite size), same as Redis. + #[arg(long = "auto-aof-rewrite-percentage", default_value_t = 100)] + pub auto_aof_rewrite_percentage: u64, + + /// Automatic AOF rewrite floor: never auto-rewrite while the total AOF + /// size is below this (Redis parity: `auto-aof-rewrite-min-size`). + /// Accepts size strings ("64mb", "1gb") or raw bytes. Default "64mb". + #[arg(long = "auto-aof-rewrite-min-size", default_value = "64mb")] + pub auto_aof_rewrite_min_size: String, + /// Max time (ms) a write may block awaiting the `appendfsync=always` /// fsync ack before the write is failed instead of parking the /// connection forever. Design-for-failure bound: a stalled disk must diff --git a/src/main.rs b/src/main.rs index 9c0f60ddc..46a8f668b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -968,40 +968,18 @@ fn main() -> anyhow::Result<()> { // Compute bind address for SO_REUSEPORT per-shard listeners (Linux io_uring path). let bind_addr = format!("{}:{}", config.bind, config.port); - // FIX-W1-4: gate BGREWRITEAOF whenever per-shard AOF is active - // (num_shards >= 2 + appendonly=yes). The original gate was too narrow: - // it required disk_offload to be enabled, missing the plain AOF case. - // Per-shard rewrite is not yet implemented (AofPoolSendError:: - // RewriteUnsupportedInPerShard); the pool already refuses the message, - // but this early gate provides a stable, documented error to operators - // BEFORE the channel send so no in-progress flag is flipped. - // Verified 2026-05-26: multi-shard BGREWRITEAOF loses ~38% of keys on - // restart. Gate lifted only when multi-part AOF replay ships (v2.0+). - // See docs/runbooks/multi-shard-aof-rewrite.md. - // [F6] When `--experimental-per-shard-rewrite` is set, leave the gate OPEN - // so BGREWRITEAOF routes to the per-shard fan-out coordinator - // (try_send_rewrite_per_shard): synchronized seq bump + single manifest - // commit across all per-shard writers, validated by - // tests/crash_matrix_per_shard_bgrewriteaof.rs. Default (flag off) keeps - // the gate closed — the shipped, crash-safe "no in-place compaction" - // behavior that avoided the historical ~38%-key-loss-on-restart. - if config.per_shard_aof_active(num_shards) { - if config.experimental_per_shard_rewrite { - tracing::warn!( - shards = num_shards, - appendonly = %config.appendonly, - "BGREWRITEAOF per-shard rewrite ENABLED (--experimental-per-shard-rewrite). \ - Per-shard fan-out compaction is active; this path is experimental." - ); - } else { - moon::command::persistence::MULTI_SHARD_AOF_REWRITE_UNSAFE - .store(true, std::sync::atomic::Ordering::Relaxed); - tracing::warn!( - shards = num_shards, - appendonly = %config.appendonly, - "BGREWRITEAOF gated: per-shard AOF layout active (see docs/runbooks/multi-shard-aof-rewrite.md). Use --shards 1, or --experimental-per-shard-rewrite to enable per-shard compaction." - ); - } + // #433: per-shard BGREWRITEAOF is the DEFAULT — the gate that refused it + // (historical ~38%-key-loss era, pre-C4 cooperative snapshot) is retired. + // The fan-out path is validated by tests/crash_matrix_per_shard_bgrewriteaof.rs + // (exact INCR recovery across a straddling rewrite + SIGKILL) and + // tests/aof_auto_rewrite.rs. `MULTI_SHARD_AOF_REWRITE_UNSAFE` is never set + // at boot anymore; the refusal branch it guards stays as dead-man code for + // tests and any future re-gate. + if config.experimental_per_shard_rewrite { + tracing::warn!( + "--experimental-per-shard-rewrite is deprecated and now a no-op: \ + per-shard BGREWRITEAOF is the default (#433)." + ); } // Create watch channel for snapshot triggers (auto-save and BGSAVE) @@ -1764,6 +1742,31 @@ fn main() -> anyhow::Result<()> { moon::shard::shared_databases::replay_mq_wal(&mut slice_inits, dir_path); } + // #433: AOF auto-rewrite monitor + INFO size statics. Init AFTER recovery + // (the just-replayed generation is the growth baseline), spawn regardless + // of percentage so `INFO persistence` sizes stay fresh; percentage 0 + // disables only the trigger. + if let Some(ref pool) = aof_pool { + moon::persistence::aof::auto_rewrite::init( + std::path::Path::new(&config.dir), + &config.appendfilename, + ); + let min_size = + ServerConfig::parse_size(&config.auto_aof_rewrite_min_size).unwrap_or_else(|| { + tracing::warn!( + "unparseable --auto-aof-rewrite-min-size {:?}; using 64mb", + config.auto_aof_rewrite_min_size + ); + 64 * 1024 * 1024 + }); + moon::persistence::aof::auto_rewrite::spawn_monitor( + pool.clone(), + shard_databases.clone(), + config.auto_aof_rewrite_percentage, + min_size, + ); + } + // All shards recovered — mark server as ready for /readyz. moon::admin::metrics_setup::set_server_ready(); // Register global ShardDatabases for MEMORY DOCTOR + Prometheus per-kind gauges. diff --git a/src/persistence/aof/auto_rewrite.rs b/src/persistence/aof/auto_rewrite.rs new file mode 100644 index 000000000..675a6443a --- /dev/null +++ b/src/persistence/aof/auto_rewrite.rs @@ -0,0 +1,275 @@ +//! #433 — automatic AOF rewrite (Redis parity: `auto-aof-rewrite-percentage` +//! / `auto-aof-rewrite-min-size`). +//! +//! The AOF is append-only: without compaction it grows with write volume, +//! not dataset size (observed in production: 4.8 GB appendonlydir over a +//! 2.43 GB dataset, ~1 GB/day), ending in the diskfull guard pausing writes. +//! This module adds the missing serverCron-equivalent: a small monitor +//! thread samples the on-disk AOF size once a second and dispatches the same +//! rewrite entry point the `BGREWRITEAOF` command uses when the growth +//! threshold is crossed. +//! +//! Semantics (mirrors Redis): +//! - trigger when `current >= min_size` AND +//! `(current - base) * 100 / max(base, 1) >= percentage` +//! - `percentage == 0` disables automatic rewrites entirely. +//! - `base` is the total AOF size right after boot recovery and after each +//! completed rewrite. +//! +//! Design-for-failure: +//! - A failed dispatch (backpressure, unsupported layout, …) arms a 60 s +//! cooldown so the monitor cannot livelock hot-retrying a rewrite that +//! keeps failing (lesson from the vector merge-backoff livelock). +//! - The monitor is skipped while a rewrite or BGSAVE is already running. +//! - Everything here is off the hot path: one directory walk per second. +//! +//! The sampled sizes double as the backing store for the `INFO persistence` +//! fields `aof_enabled` / `aof_base_size` / `aof_current_size` (#432 fixed +//! the hardcoded zeros). + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, OnceLock}; +use tracing::{info, warn}; + +/// Whether AOF persistence is enabled at all (set once at boot when the +/// writer pool is created). Backs `INFO persistence` `aof_enabled` (#432). +pub static AOF_ENABLED: AtomicBool = AtomicBool::new(false); + +/// Total on-disk AOF size (bytes) at boot / after the last completed +/// rewrite. Backs `INFO persistence` `aof_base_size`. +pub static AOF_BASE_SIZE: AtomicU64 = AtomicU64::new(0); + +/// Most recently sampled total on-disk AOF size (bytes). Backs +/// `INFO persistence` `aof_current_size`. Refreshed by the monitor tick and +/// by `refresh_current_size` (INFO reads between ticks / monitor disabled). +pub static AOF_CURRENT_SIZE: AtomicU64 = AtomicU64::new(0); + +/// The `appendonlydir` root the sizes are measured from. +static AOF_DIR: OnceLock = OnceLock::new(); + +/// Cooldown after a failed auto-rewrite dispatch. One minute mirrors the +/// "don't hot-retry a deterministic failure" backoff floor used elsewhere. +const FAILED_DISPATCH_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(60); + +/// Monitor sampling cadence. +const TICK: std::time::Duration = std::time::Duration::from_secs(1); + +/// Legacy single-file AOF path (`/`), used by the tokio +/// TopLevel writer, which appends to one flat file instead of the +/// `appendonlydir` manifest layout the monoio writers use. Both are measured; +/// whichever exists contributes (they never coexist for one server). +static AOF_LEGACY_FILE: OnceLock = OnceLock::new(); + +/// Record the AOF locations (`/appendonlydir` + the legacy +/// `/` flat file) and initialize the base / current +/// sizes from what recovery just replayed. Call once at boot, after AOF +/// recovery, when appendonly is enabled. +pub fn init(persistence_dir: &Path, appendfilename: &str) { + let _ = AOF_DIR.set(persistence_dir.join("appendonlydir")); + let _ = AOF_LEGACY_FILE.set(persistence_dir.join(appendfilename)); + AOF_ENABLED.store(true, Ordering::Relaxed); + record_base_size(); +} + +/// Total on-disk AOF size: every file under the manifest root (base RDBs + +/// incr logs + manifest, all generations, both manifest layouts) PLUS the +/// legacy flat file when present. 0 when uninitialized. +pub fn measure_total_size() -> u64 { + fn walk(p: &Path, total: &mut u64) { + let Ok(entries) = std::fs::read_dir(p) else { + return; + }; + for e in entries.flatten() { + let path = e.path(); + if path.is_dir() { + walk(&path, total); + } else if let Ok(md) = e.metadata() { + *total += md.len(); + } + } + } + let mut total = 0; + if let Some(dir) = AOF_DIR.get() { + walk(dir, &mut total); + } + if let Some(file) = AOF_LEGACY_FILE.get() + && let Ok(md) = std::fs::metadata(file) + { + total += md.len(); + } + total +} + +/// Re-measure and store BOTH base and current size — the AOF was just +/// compacted (or just recovered), so "now" is the new growth baseline. +pub fn record_base_size() { + let size = measure_total_size(); + AOF_BASE_SIZE.store(size, Ordering::Relaxed); + AOF_CURRENT_SIZE.store(size, Ordering::Relaxed); +} + +/// Re-measure the current size only (INFO freshness between monitor ticks). +pub fn refresh_current_size() -> u64 { + let size = measure_total_size(); + AOF_CURRENT_SIZE.store(size, Ordering::Relaxed); + size +} + +/// The Redis auto-rewrite predicate. Pure so the boundary math is unit +/// tested without a filesystem. +pub fn should_trigger(current: u64, base: u64, percentage: u64, min_size: u64) -> bool { + if percentage == 0 || current < min_size { + return false; + } + // Redis uses max(base, 1) so a never-rewritten (base 0) AOF still + // triggers once it crosses min_size. + let base = base.max(1); + current.saturating_sub(base).saturating_mul(100) / base >= percentage +} + +/// Spawn the auto-rewrite monitor thread. `percentage == 0` still spawns the +/// sampler (INFO size freshness) but never dispatches a rewrite. +/// +/// Dispatches through [`crate::command::persistence::bgrewriteaof_start_sharded`], +/// the exact entry the `BGREWRITEAOF` command uses — CAS on the in-progress +/// flag, per-shard fan-out vs TopLevel routing, and error mapping included. +pub fn spawn_monitor( + pool: Arc, + shard_databases: Arc, + percentage: u64, + min_size: u64, +) { + let spawned = std::thread::Builder::new() + .name("aof-auto-rewrite".to_string()) + .spawn(move || { + monitor_loop(&pool, &shard_databases, percentage, min_size); + }); + if let Err(e) = spawned { + // Non-fatal: manual BGREWRITEAOF still works; sizes go stale. + warn!("aof-auto-rewrite monitor failed to spawn: {e}"); + } +} + +fn monitor_loop( + pool: &super::AofWriterPool, + shard_databases: &Arc, + percentage: u64, + min_size: u64, +) { + use crate::command::persistence::{AOF_REWRITE_IN_PROGRESS, SAVE_IN_PROGRESS}; + use crate::protocol::Frame; + + let mut cooldown_until = std::time::Instant::now(); + let mut saw_in_progress = false; + info!( + "aof-auto-rewrite monitor started (percentage={}%, min_size={} bytes)", + percentage, min_size + ); + loop { + std::thread::sleep(TICK); + + let in_progress = AOF_REWRITE_IN_PROGRESS.load(Ordering::SeqCst); + let current = refresh_current_size(); + + // A rewrite completed since the last tick (ours or a manual + // BGREWRITEAOF): the compacted size is the new growth baseline. + // The `current < base` clause catches rewrites so fast that both + // the set and the clear of the flag fell between two ticks — the + // only other way the AOF shrinks is a rewrite committing + pruning. + let completed_since_last_tick = saw_in_progress && !in_progress; + let shrunk_below_base = current < AOF_BASE_SIZE.load(Ordering::Relaxed); + saw_in_progress = in_progress; + if completed_since_last_tick || (!in_progress && shrunk_below_base) { + record_base_size(); + continue; + } + + if percentage == 0 + || in_progress + || SAVE_IN_PROGRESS.load(Ordering::SeqCst) + || std::time::Instant::now() < cooldown_until + { + continue; + } + + let base = AOF_BASE_SIZE.load(Ordering::Relaxed); + if !should_trigger(current, base, percentage, min_size) { + continue; + } + + info!( + "aof-auto-rewrite: triggering BGREWRITEAOF (current={} base={} \ + growth>={}%, min_size={})", + current, base, percentage, min_size + ); + match crate::command::persistence::bgrewriteaof_start_sharded(pool, shard_databases.clone()) + { + Frame::Error(e) => { + warn!( + "aof-auto-rewrite: dispatch failed ({}); retrying in {:?}", + String::from_utf8_lossy(&e), + FAILED_DISPATCH_COOLDOWN + ); + cooldown_until = std::time::Instant::now() + FAILED_DISPATCH_COOLDOWN; + } + _ => { + // Started. Wait for completion here (bounded) so the rebase + // is deterministic even when the whole rewrite fits inside + // one tick; the transition/shrink detection above is the + // fallback for manual rewrites. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(300); + while AOF_REWRITE_IN_PROGRESS.load(Ordering::SeqCst) + && std::time::Instant::now() < deadline + { + std::thread::sleep(std::time::Duration::from_millis(100)); + } + record_base_size(); + saw_in_progress = false; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::should_trigger; + + #[test] + fn percentage_zero_never_triggers() { + assert!(!should_trigger(u64::MAX, 0, 0, 0)); + assert!(!should_trigger(1 << 30, 64, 0, 1024)); + } + + #[test] + fn min_size_floor_holds() { + // 100% growth but below the floor: no trigger. + assert!(!should_trigger(1000, 500, 100, 4096)); + // At the floor: trigger. + assert!(should_trigger(4096, 2048, 100, 4096)); + } + + #[test] + fn growth_boundary_is_inclusive() { + // base 1000, +100% => current 2000 triggers, 1999 does not. + assert!(should_trigger(2000, 1000, 100, 0)); + assert!(!should_trigger(1999, 1000, 100, 0)); + // 50% threshold. + assert!(should_trigger(1500, 1000, 50, 0)); + assert!(!should_trigger(1499, 1000, 50, 0)); + } + + #[test] + fn zero_base_uses_one_like_redis() { + // Never-rewritten AOF: any current >= min_size is astronomically + // over any percentage of base=1. + assert!(should_trigger(4096, 0, 100, 4096)); + assert!(!should_trigger(4095, 0, 100, 4096)); + } + + #[test] + fn no_overflow_at_extremes() { + assert!(should_trigger(u64::MAX, 1, 100, 0)); + assert!(!should_trigger(0, u64::MAX, 100, 0)); + } +} diff --git a/src/persistence/aof/mod.rs b/src/persistence/aof/mod.rs index c214f3752..220bea070 100644 --- a/src/persistence/aof/mod.rs +++ b/src/persistence/aof/mod.rs @@ -487,6 +487,9 @@ pub const DEFAULT_AOF_FSYNC_TIMEOUT: Duration = Duration::from_millis(2000); // ── Submodule decomposition (refactor: aof.rs 4379 lines -> directory module) ── // Codec (serialize_command/replay_aof) stays in this parent so children reach it // via `use super::*`. AofWriterPool, writer tasks, and rewrite paths move out. +/// #433 — automatic AOF rewrite monitor (Redis `auto-aof-rewrite-*` parity) +/// plus the INFO-persistence size/enabled statics (#432). +pub mod auto_rewrite; /// Group-commit batching seam (coalesce concurrent pending writes into one /// fsync under `appendfsync=always`). `pub` so the §4 red suite can pin the pure /// seam (collect/commit) against the public API. diff --git a/tests/aof_auto_rewrite.rs b/tests/aof_auto_rewrite.rs new file mode 100644 index 000000000..cd05d551c --- /dev/null +++ b/tests/aof_auto_rewrite.rs @@ -0,0 +1,397 @@ +//! #433: automatic AOF rewrite (`auto-aof-rewrite-percentage` / +//! `auto-aof-rewrite-min-size`, Redis parity) + un-gated multi-shard +//! BGREWRITEAOF. +//! +//! The AOF is append-only; before #433 nothing ever compacted it — it grew +//! with write volume, not dataset size (observed: 4.8 GB appendonlydir for a +//! 2.43 GB dataset, +1 GB/day), ending in `MOONERR diskfull`. And on the +//! default multi-shard config the manual escape hatch was gated off too +//! (`BGREWRITEAOF` refused unless `--experimental-per-shard-rewrite`). +//! +//! These tests pin the new contract: +//! 1. Multi-shard `BGREWRITEAOF` works WITHOUT the experimental flag (the +//! per-shard fan-out is the default; crash matrix green). +//! 2. The AOF rewrites ITSELF when it grows `percentage`% over its +//! post-rewrite size and is at least `min-size` bytes — no operator +//! action, observable as a seq>1 base file replacing the old generation. +//! 3. `auto-aof-rewrite-percentage 0` disables the trigger (Redis parity). +//! 4. Exactness: an auto rewrite must not drop or double-apply acked +//! INCRs across the rewrite boundary (SIGKILL + recovery, same +//! non-idempotent-counter technique as the crash matrix). +//! 5. `INFO persistence` reports real `aof_enabled` / `aof_base_size` / +//! `aof_current_size` instead of hardcoded zeros (#432). +//! +//! Spawns real server binaries; needs `redis-cli` on PATH. Every server gets +//! `--disk-free-min-pct 0` (crash-harness convention — see +//! gotcha-diskfull-guard-gutted-crash-tests) and every INCR reply is parsed +//! strictly so an error reply fails loudly instead of under-counting. + +mod common; + +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +fn unique_dir(suffix: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + std::env::temp_dir().join(format!( + "moon-aof-auto-rewrite-{}-{}-{}", + std::process::id(), + suffix, + nanos + )) +} + +/// Spawn moon with the given shard count and extra flags. Deliberately does +/// NOT pass `--experimental-per-shard-rewrite`: the un-gated default is part +/// of the contract under test. +fn start_moon(port: u16, dir: &std::path::Path, shards: usize, extra: &[&str]) -> Child { + let port_s = port.to_string(); + let shards_s = shards.to_string(); + let mut args: Vec<&str> = vec![ + "--port", + &port_s, + "--shards", + &shards_s, + "--appendonly", + "yes", + "--appendfsync", + "everysec", + "--disk-free-min-pct", + "0", + ]; + args.extend_from_slice(extra); + let mut cmd = Command::new(common::find_moon_binary()); + cmd.args(&args).arg("--dir").arg(dir); + cmd.stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log")) + .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log")) + .spawn() + .expect("spawn moon (cargo build first; MOON_BIN to override)") +} + +fn spawn_listening(dir: &std::path::Path, shards: usize, extra: &[&str]) -> (Child, u16) { + common::spawn_listening(|port| start_moon(port, dir, shards, extra)) +} + +fn cli(port: u16, args: &[&str]) -> String { + let mut full = vec!["-p".to_string(), port.to_string()]; + full.extend(args.iter().map(|s| s.to_string())); + let out = Command::new("redis-cli") + .args(&full) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .expect("redis-cli"); + String::from_utf8_lossy(&out.stdout).trim().to_string() +} + +/// Strict INCR: panics on a non-numeric reply so error replies (diskfull, +/// readonly, …) can never masquerade as lost writes. +fn incr(port: u16, key: &str) -> i64 { + let reply = cli(port, &["INCR", key]); + reply.parse().unwrap_or_else(|_| { + panic!("INCR {key} not acked with a number — harness cannot count it: {reply:?}") + }) +} + +fn sigkill(child: &mut Child) { + #[cfg(unix)] + unsafe { + libc::kill(child.id() as i32, libc::SIGKILL); + } + #[cfg(not(unix))] + let _ = child.kill(); + let _ = child.wait(); +} + +/// True when a base RDB with seq > `min_seq_exclusive` exists anywhere under +/// appendonlydir (covers both the TopLevel `appendonlydir/moon.aof.N.base.rdb` +/// and PerShard `appendonlydir/shard-K/moon.aof.N.base.rdb` layouts). +fn max_base_seq(dir: &std::path::Path) -> u64 { + fn scan(p: &std::path::Path, max: &mut u64) { + let Ok(entries) = std::fs::read_dir(p) else { + return; + }; + for e in entries.flatten() { + let path = e.path(); + if path.is_dir() { + scan(&path, max); + } else if let Some(name) = path.file_name().and_then(|n| n.to_str()) + && let Some(rest) = name.strip_prefix("moon.aof.") + && let Some(seq_str) = rest.strip_suffix(".base.rdb") + && let Ok(seq) = seq_str.parse::() + { + *max = (*max).max(seq); + } + } + } + let mut max = 0; + scan(&dir.join("appendonlydir"), &mut max); + max +} + +fn wait_for_base_seq_above(dir: &std::path::Path, floor: u64, timeout: Duration) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if max_base_seq(dir) > floor { + return true; + } + std::thread::sleep(Duration::from_millis(200)); + } + false +} + +/// Layout-aware "a rewrite compacted the AOF" tracker. Manifest layouts +/// (monoio: both shard counts; tokio: shards>=2) advance to a seq>1 base +/// file — visible forever after. The tokio TopLevel writer instead compacts +/// its legacy flat `appendonly.aof` IN PLACE, and the file starts regrowing +/// immediately — so the shrink-below-high-water signal is only observable +/// while it happens. Callers must therefore `sample()` DURING the write +/// stream too, not only after it. +struct CompactionTracker { + legacy: std::path::PathBuf, + high_water: u64, + seen: bool, +} + +impl CompactionTracker { + fn new(dir: &std::path::Path) -> Self { + Self { + legacy: dir.join("appendonly.aof"), + high_water: 0, + seen: false, + } + } + + fn sample(&mut self, dir: &std::path::Path) -> bool { + if self.seen { + return true; + } + if max_base_seq(dir) > 1 { + self.seen = true; + return true; + } + if let Ok(md) = std::fs::metadata(&self.legacy) { + let len = md.len(); + if self.high_water > 1024 && len + 1024 < self.high_water { + // Shrank well below the high-water mark: in-place rewrite. + self.seen = true; + return true; + } + self.high_water = self.high_water.max(len); + } + false + } + + fn wait(&mut self, dir: &std::path::Path, timeout: Duration) -> bool { + let deadline = std::time::Instant::now() + timeout; + while std::time::Instant::now() < deadline { + if self.sample(dir) { + return true; + } + std::thread::sleep(Duration::from_millis(200)); + } + false + } +} + +fn wait_for_compaction(dir: &std::path::Path, timeout: Duration) -> bool { + CompactionTracker::new(dir).wait(dir, timeout) +} + +/// Contract 1: multi-shard BGREWRITEAOF works by DEFAULT (no experimental +/// flag) and compacts — plus exactness across SIGKILL+recovery. +#[test] +#[ignore] // Spawns real binaries + SIGKILL; run explicitly (crash-suite convention). +fn manual_bgrewriteaof_ungated_on_multi_shard() { + const N: i64 = 200; + let dir = unique_dir("ungated"); + std::fs::create_dir_all(&dir).unwrap(); + let (mut child, port) = spawn_listening(&dir, 2, &[]); + + for _ in 0..N { + incr(port, "cnt:{a}"); + incr(port, "cnt:{b}"); + } + let reply = cli(port, &["BGREWRITEAOF"]); + assert!( + reply.contains("started"), + "multi-shard BGREWRITEAOF must be un-gated by default, got: {reply:?}" + ); + assert!( + wait_for_base_seq_above(&dir, 1, Duration::from_secs(10)), + "no compacted (seq>1) base appeared after manual BGREWRITEAOF" + ); + // Post-rewrite writes + quiesce past the everysec window, then crash. + for _ in 0..50 { + incr(port, "cnt:{a}"); + } + std::thread::sleep(Duration::from_millis(2000)); + sigkill(&mut child); + + let (mut child2, port2) = spawn_listening(&dir, 2, &[]); + assert_eq!(cli(port2, &["GET", "cnt:{a}"]), (N + 50).to_string()); + assert_eq!(cli(port2, &["GET", "cnt:{b}"]), N.to_string()); + sigkill(&mut child2); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Contract 2 + 4 (multi-shard): the AOF auto-rewrites once it exceeds the +/// growth threshold, with INCR-exact recovery across SIGKILL. +#[test] +#[ignore] // Spawns real binaries + SIGKILL; run explicitly (crash-suite convention). +fn auto_rewrite_triggers_on_growth_multi_shard() { + const N: i64 = 400; + let dir = unique_dir("auto2"); + std::fs::create_dir_all(&dir).unwrap(); + // Tiny thresholds so ~800 INCR records (~30 bytes each) cross quickly. + let (mut child, port) = spawn_listening( + &dir, + 2, + &[ + "--auto-aof-rewrite-min-size", + "4096", + "--auto-aof-rewrite-percentage", + "50", + ], + ); + + for _ in 0..N { + incr(port, "cnt:{a}"); + incr(port, "cnt:{b}"); + } + assert!( + wait_for_base_seq_above(&dir, 1, Duration::from_secs(20)), + "auto rewrite never fired: no seq>1 base under {} (incr grew past \ + min-size+percentage but nothing compacted)", + dir.display() + ); + // Keep writing across/after the rewrite, quiesce, crash, recover exact. + for _ in 0..100 { + incr(port, "cnt:{a}"); + } + std::thread::sleep(Duration::from_millis(2000)); + sigkill(&mut child); + + let (mut child2, port2) = spawn_listening(&dir, 2, &[]); + assert_eq!( + cli(port2, &["GET", "cnt:{a}"]), + (N + 100).to_string(), + "auto rewrite dropped or double-applied acked INCRs for cnt:{{a}}" + ); + assert_eq!(cli(port2, &["GET", "cnt:{b}"]), N.to_string()); + sigkill(&mut child2); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Contract 2 (single-shard / TopLevel layout): same trigger, legacy rewrite +/// path. +#[test] +#[ignore] // Spawns real binaries + SIGKILL; run explicitly (crash-suite convention). +fn auto_rewrite_triggers_on_growth_single_shard() { + const N: i64 = 500; + let dir = unique_dir("auto1"); + std::fs::create_dir_all(&dir).unwrap(); + let (mut child, port) = spawn_listening( + &dir, + 1, + &[ + "--auto-aof-rewrite-min-size", + "4096", + "--auto-aof-rewrite-percentage", + "50", + ], + ); + // Sample DURING the stream: the tokio TopLevel layout compacts its flat + // file in place mid-stream and regrows it — a post-hoc poll misses it. + let mut tracker = CompactionTracker::new(&dir); + for i in 0..N { + incr(port, "cnt:solo"); + if i % 10 == 0 { + tracker.sample(&dir); + } + } + assert!( + tracker.wait(&dir, Duration::from_secs(20)), + "auto rewrite never fired on the shards=1 layout (no seq>1 base and \ + no in-place shrink of appendonly.aof)" + ); + std::thread::sleep(Duration::from_millis(2000)); + sigkill(&mut child); + let (mut child2, port2) = spawn_listening(&dir, 1, &[]); + assert_eq!(cli(port2, &["GET", "cnt:solo"]), N.to_string()); + sigkill(&mut child2); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Contract 3: percentage 0 disables the trigger entirely. +#[test] +#[ignore] // Spawns real binaries; run explicitly (crash-suite convention). +fn auto_rewrite_percentage_zero_disables() { + let dir = unique_dir("disabled"); + std::fs::create_dir_all(&dir).unwrap(); + let (mut child, port) = spawn_listening( + &dir, + 2, + &[ + "--auto-aof-rewrite-min-size", + "1024", + "--auto-aof-rewrite-percentage", + "0", + ], + ); + for _ in 0..300 { + incr(port, "cnt:{a}"); + } + // Well past several monitor ticks: nothing may compact. + std::thread::sleep(Duration::from_millis(4000)); + assert_eq!( + max_base_seq(&dir), + 1, + "auto-aof-rewrite-percentage 0 must disable automatic rewrites" + ); + sigkill(&mut child); + let _ = std::fs::remove_dir_all(&dir); +} + +/// Contract 5 (#432): INFO persistence must report the real AOF state. +#[test] +#[ignore] // Spawns real binaries; run explicitly (crash-suite convention). +fn info_persistence_reports_real_aof_fields() { + let dir = unique_dir("info"); + std::fs::create_dir_all(&dir).unwrap(); + let (mut child, port) = spawn_listening(&dir, 2, &[]); + for _ in 0..50 { + incr(port, "cnt:{a}"); + } + // everysec: give the writer a flush window before sampling sizes. + std::thread::sleep(Duration::from_millis(1500)); + let info = cli(port, &["INFO", "persistence"]); + assert!( + info.contains("aof_enabled:1"), + "appendonly=yes must report aof_enabled:1 (#432), got:\n{info}" + ); + assert!( + info.contains("aof_rewrite_in_progress:0"), + "no rewrite is running, got:\n{info}" + ); + let current = info + .lines() + .find_map(|l| l.strip_prefix("aof_current_size:")) + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or_else(|| panic!("missing/unparseable aof_current_size in:\n{info}")); + let base = info + .lines() + .find_map(|l| l.strip_prefix("aof_base_size:")) + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or_else(|| panic!("missing/unparseable aof_base_size in:\n{info}")); + assert!( + current > base, + "50 INCRs were appended, aof_current_size ({current}) must exceed \ + aof_base_size ({base})" + ); + sigkill(&mut child); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/tests/crash_matrix_per_shard_bgrewriteaof.rs b/tests/crash_matrix_per_shard_bgrewriteaof.rs index 31dbd2d48..5f0a18d72 100644 --- a/tests/crash_matrix_per_shard_bgrewriteaof.rs +++ b/tests/crash_matrix_per_shard_bgrewriteaof.rs @@ -20,9 +20,10 @@ //! cargo build --release //! cargo test --release --test crash_matrix_per_shard_bgrewriteaof -- --ignored //! -//! Requires: built release binary (default features = runtime-monoio) and -//! `redis-cli` on PATH. Per-shard BGREWRITEAOF is monoio-only (the fold uses -//! synchronous std::fs IO); the tokio build refuses it at the command handler. +//! Requires: built release binary and `redis-cli` on PATH. Both runtimes run +//! the same synchronous fold (`do_rewrite_per_shard`): monoio writers inline +//! it; tokio writers convert their file handle to `std::fs` for its duration +//! (dedicated block_on_local threads, so blocking is safe). #![cfg(any(feature = "runtime-monoio", feature = "runtime-tokio"))] @@ -66,6 +67,14 @@ fn start_moon(port: u16, dir: &std::path::Path) -> Child { // F6: open the gate so BGREWRITEAOF routes to the per-shard // fan-out coordinator instead of the refusal error. "--experimental-per-shard-rewrite", + // Diskfull guard guts crash tests: on hosts hovering near the 5% + // free-space threshold the guard rejects mid-stream INCRs with + // MOONERR diskfull, and recovery then "loses" writes that were in + // fact never acked (observed: host at 3.9% free → recovered 191 + // of 500 with the other 309 rejected). Crash harnesses always + // disable it. + "--disk-free-min-pct", + "0", "--dir", ]) .arg(dir) @@ -89,7 +98,10 @@ fn wait_for_port(port: u16) { panic!("moon did not start within 8s on port {}", port); } -/// One `INCR key`, returning the new value (or -1 on failure). +/// One `INCR key`, returning the new value. Panics on a non-numeric reply: +/// a silently swallowed error (e.g. `MOONERR diskfull` pausing writes +/// mid-stream) makes the exact-count recovery assertion report phantom +/// "lost writes" that were in fact never acked. fn redis_incr(port: u16, key: &str) -> i64 { let out = Command::new("redis-cli") .args(["-p", &port.to_string(), "INCR", key]) @@ -97,10 +109,10 @@ fn redis_incr(port: u16, key: &str) -> i64 { .stderr(Stdio::piped()) .output() .expect("redis-cli INCR"); - String::from_utf8_lossy(&out.stdout) - .trim() - .parse() - .unwrap_or(-1) + let reply = String::from_utf8_lossy(&out.stdout).trim().to_string(); + reply.parse().unwrap_or_else(|_| { + panic!("INCR {key} was not acked with a number — harness cannot count it: {reply:?}") + }) } fn redis_get_i64(port: u16, key: &str) -> i64 { diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs index 76ea3f859..3c5a069c9 100644 --- a/tests/mq_integration.rs +++ b/tests/mq_integration.rs @@ -52,6 +52,8 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) { appendfsync: "everysec".to_string(), aof_fsync_timeout_ms: 2000, experimental_per_shard_rewrite: false, + auto_aof_rewrite_percentage: 100, + auto_aof_rewrite_min_size: "64mb".to_string(), save: None, dir: ".".to_string(), dbfilename: "dump.rdb".to_string(), diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 540b85a1e..d316e4e78 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -62,6 +62,8 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can appendonly: appendonly.clone(), unsafe_multishard_aof: false, experimental_per_shard_rewrite: false, + auto_aof_rewrite_percentage: 100, + auto_aof_rewrite_min_size: "64mb".to_string(), appendfsync: "everysec".to_string(), aof_fsync_timeout_ms: 2000, save: None, diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs index 16b2b0e48..d3ba27456 100644 --- a/tests/workspace_integration.rs +++ b/tests/workspace_integration.rs @@ -43,6 +43,8 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) { appendfsync: "everysec".to_string(), aof_fsync_timeout_ms: 2000, experimental_per_shard_rewrite: false, + auto_aof_rewrite_percentage: 100, + auto_aof_rewrite_min_size: "64mb".to_string(), save: None, dir: ".".to_string(), dbfilename: "dump.rdb".to_string(), @@ -291,6 +293,8 @@ async fn start_workspace_server_with_auth( appendfsync: "everysec".to_string(), aof_fsync_timeout_ms: 2000, experimental_per_shard_rewrite: false, + auto_aof_rewrite_percentage: 100, + auto_aof_rewrite_min_size: "64mb".to_string(), save: None, dir: ".".to_string(), dbfilename: "dump.rdb".to_string(),