From 7db49655721fd0a96864d56bbe9de84b6999ea31 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 16:11:05 +0700 Subject: [PATCH 1/4] =?UTF-8?q?feat(shard):=20add=20mem=5Fmonitor=20?= =?UTF-8?q?=E2=80=94=20proactive=20RSS=20watchdog=20core=20+=20INFO=20wiri?= =?UTF-8?q?ng?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding #6 from tmp/RSS-CPU-REVIEW.md (wave 3 of the RSS/CPU/OOM train). Moon has a diskfull guard (`MOONERR diskfull` at <5% free) but no memory analogue — when --maxmemory is unset (or too high), RSS can grow until the kernel OOM-killer picks Moon. This adds mem_monitor, mirroring disk_monitor's struct + hysteresis state machine + inject() test backdoor + OnceLock global singleton pattern exactly, inverted for direction (high RSS is bad, not low free space): writes pause once rss% >= mem_full_pct and resume only once rss% <= mem_full_pct - 5 (hysteresis prevents flapping). Fires on ACTUAL RSS (via the existing zero-alloc get_rss_bytes()) vs the DETECTED system/cgroup memory limit (detect_memory_limit_bytes, now pub(crate) so mem_monitor can call it), not the configured --maxmemory (which can be an unconfigured 0). limit_bytes is resolved ONCE at init_global time and held fixed — unlike disk_monitor's per-poll statvfs, the memory-limit probe can shell out on macOS and must never run on the hot poll path. A hidden MOON_MEM_LIMIT_BYTES env override (documented in the module header, same rationale as the MOON_XSHARD_* bench-diagnostic knobs) lets tests engage the guard deterministically without controlling real process RSS. New --mem-full-pct CLI flag (default 95, 0 = disabled) alongside --disk-free-min-pct in config.rs. New INFO fields reclamation_mem_rss_bytes / reclamation_mem_watchdog_active in info_reclamation.rs, OR'd into the existing reclamation_write_stall_active aggregate. Red/green: 15 mem_monitor unit tests (threshold, INVERTED exact-threshold pause, hysteresis resume/flap, disabled via pause_pct=0 or limit_bytes=0, u8-wrap clamp on rss >> limit, initial-optimistic, Send+Sync, real-process smoke) + 1 new info_reclamation wiring test (default false, OR-merge into write_stall_active). author: Tin Dang --- src/command/info_reclamation.rs | 54 +++- src/config.rs | 25 +- src/shard/mem_monitor.rs | 500 ++++++++++++++++++++++++++++++++ src/shard/mod.rs | 3 + 4 files changed, 577 insertions(+), 5 deletions(-) create mode 100644 src/shard/mem_monitor.rs diff --git a/src/command/info_reclamation.rs b/src/command/info_reclamation.rs index 96259110c..e0d78111c 100644 --- a/src/command/info_reclamation.rs +++ b/src/command/info_reclamation.rs @@ -43,6 +43,15 @@ pub static RECL_WRITE_STALL_ACTIVE: AtomicU64 = AtomicU64::new(0); /// MA1 sets this; the INFO `write_stall_active` field reflects its OR with RECL_WRITE_STALL_ACTIVE. pub static RECL_SEGMENT_STALL_ACTIVE: AtomicU64 = AtomicU64::new(0); +/// Latest measured process RSS in bytes. Wave 3 (`mem_monitor`) owns this; +/// emits 0 until the monitor's first poll. +pub static RECL_MEM_RSS_BYTES: AtomicU64 = AtomicU64::new(0); + +/// 1 when the RSS memory watchdog (Wave 3, `mem_monitor`) has paused writes +/// due to memory pressure, 0 otherwise. The INFO `write_stall_active` field +/// ORs this in alongside RECL_WRITE_STALL_ACTIVE and RECL_SEGMENT_STALL_ACTIVE. +pub static RECL_MEM_WATCHDOG_ACTIVE: AtomicU64 = AtomicU64::new(0); + /// Write-stall threshold stored as tenths-of-percent (e.g. 950 = 95.0%). /// TODO(P10→Wave2): wire from config flag --disk-free-min-pct (MA12). pub static RECL_WRITE_STALL_THRESHOLD_PCT_X10: AtomicU64 = AtomicU64::new(950); @@ -172,6 +181,16 @@ pub fn write_reclamation_section(buf: &mut String) { RECL_DISK_FREE_BYTES.load(Ordering::Relaxed) ); + // -- Memory (Wave 3 RSS watchdog) -- + let mem_watchdog_active = RECL_MEM_WATCHDOG_ACTIVE.load(Ordering::Relaxed) != 0; + let _ = write!( + buf, + "reclamation_mem_rss_bytes:{}\r\n\ + reclamation_mem_watchdog_active:{}\r\n", + RECL_MEM_RSS_BYTES.load(Ordering::Relaxed), + if mem_watchdog_active { "true" } else { "false" }, + ); + // -- WAL -- let _ = write!( buf, @@ -181,9 +200,11 @@ pub fn write_reclamation_section(buf: &mut String) { RECL_WAL_SEGMENTS.load(Ordering::Relaxed) ); - // -- Write stall: OR of disk-pressure (MA12) and segment-backlog (MA1) bits -- + // -- Write stall: OR of disk-pressure (MA12), segment-backlog (MA1), and + // RSS memory pressure (Wave 3) bits -- let stall_active = RECL_WRITE_STALL_ACTIVE.load(Ordering::Relaxed) - | RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed); + | RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed) + | RECL_MEM_WATCHDOG_ACTIVE.load(Ordering::Relaxed); // Threshold stored as tenths-of-percent (e.g. 950 → "95.0") let threshold_x10 = RECL_WRITE_STALL_THRESHOLD_PCT_X10.load(Ordering::Relaxed); let _ = write!( @@ -339,6 +360,8 @@ mod tests { let required_fields: &[&str] = &[ "reclamation_disk_free_bytes:", + "reclamation_mem_rss_bytes:", + "reclamation_mem_watchdog_active:", "reclamation_wal_bytes:", "reclamation_wal_segments:", "reclamation_write_stall_active:", @@ -410,6 +433,33 @@ mod tests { ); } + /// Default sentinel for the new Wave-3 `mem_watchdog_active` field must + /// be `false`, and setting `RECL_MEM_WATCHDOG_ACTIVE` must both flip its + /// own field AND OR into the aggregate `write_stall_active` — mirrors + /// `info_reclamation_wal_bytes_wirable`'s store/assert/restore shape. + #[test] + fn info_reclamation_mem_watchdog_wirable() { + let mut buf = String::new(); + write_reclamation_section(&mut buf); + assert!( + buf.contains("reclamation_mem_watchdog_active:false\r\n"), + "default mem_watchdog_active must be false" + ); + + RECL_MEM_WATCHDOG_ACTIVE.store(1, Ordering::Relaxed); + let mut buf2 = String::new(); + write_reclamation_section(&mut buf2); + RECL_MEM_WATCHDOG_ACTIVE.store(0, Ordering::Relaxed); // restore default + assert!( + buf2.contains("reclamation_mem_watchdog_active:true\r\n"), + "RECL_MEM_WATCHDOG_ACTIVE store must be visible in its own field" + ); + assert!( + buf2.contains("reclamation_write_stall_active:true\r\n"), + "RECL_MEM_WATCHDOG_ACTIVE must OR into the aggregate write_stall_active" + ); + } + /// Default `delete_pending_visible_lsn` must be `-1` (no-data sentinel). #[test] fn info_reclamation_delete_pending_lsn_default_minus_one() { diff --git a/src/config.rs b/src/config.rs index f93e60e20..a224f73c9 100644 --- a/src/config.rs +++ b/src/config.rs @@ -467,6 +467,25 @@ pub struct ServerConfig { #[arg(long = "disk-free-min-pct", default_value_t = 5, value_parser = clap::value_parser!(u8).range(0..=95))] pub disk_free_min_pct: u8, + // ── Wave 3: proactive RSS memory watchdog ("mem-full guard") ─────────── + /// Pause writes when process RSS crosses this percentage of the detected + /// system/cgroup memory limit. + /// + /// This is the memory analogue of `--disk-free-min-pct` (MA12): it fires + /// on the ACTUAL RSS vs the detected limit (`detect_memory_limit_bytes`), + /// not on the configured `--maxmemory` (which can be an unconfigured 0). + /// The direction is INVERTED vs the disk guard: high RSS is bad, so + /// writes pause once RSS% >= `mem_full_pct` and resume only once + /// RSS% <= `mem_full_pct - 5` (hysteresis, prevents flapping). + /// + /// Read-only commands are never blocked. Like the diskfull guard, + /// DEL/UNLINK/EXPIRE/FLUSHALL are write-flagged and are blocked too while + /// paused — the same accepted trade-off as MA12; no allowlist. + /// + /// Set to 0 to disable the monitor entirely. + #[arg(long = "mem-full-pct", default_value_t = 95, value_parser = clap::value_parser!(u8).range(0..=100))] + pub mem_full_pct: u8, + // ── P3: MVCC committed-set prune margin ──────────────────────────────── /// Number of LSN units to keep in the MVCC committed treemap above the /// oldest active snapshot watermark before pruning entries below. @@ -1004,7 +1023,7 @@ fn parse_cgroup_mem_max(contents: &str) -> Option { /// (v2 then v1) and host RAM. Linux-only; returns `None` elsewhere or when /// nothing is readable (the guardrail then fails open with a warning). #[cfg(target_os = "linux")] -fn detect_memory_limit_bytes() -> Option { +pub(crate) fn detect_memory_limit_bytes() -> Option { let host = std::fs::read_to_string("/proc/meminfo") .ok() .and_then(|c| parse_meminfo_memtotal(&c)); @@ -1029,7 +1048,7 @@ fn detect_memory_limit_bytes() -> Option { /// unsafe blocks; it runs once at startup so the fork cost is irrelevant. /// No container limit concept applies on macOS, so host RAM is the limit. #[cfg(target_os = "macos")] -fn detect_memory_limit_bytes() -> Option { +pub(crate) fn detect_memory_limit_bytes() -> Option { let out = std::process::Command::new("sysctl") .args(["-n", "hw.memsize"]) .output() @@ -1044,7 +1063,7 @@ fn detect_memory_limit_bytes() -> Option { /// guardrail is skipped (operator sets `--maxmemory` explicitly). Production /// targets Linux/macOS per the platform policy. #[cfg(not(any(target_os = "linux", target_os = "macos")))] -fn detect_memory_limit_bytes() -> Option { +pub(crate) fn detect_memory_limit_bytes() -> Option { None } diff --git a/src/shard/mem_monitor.rs b/src/shard/mem_monitor.rs new file mode 100644 index 000000000..22803e07a --- /dev/null +++ b/src/shard/mem_monitor.rs @@ -0,0 +1,500 @@ +//! Proactive RSS memory watchdog with write-pause / hysteresis. +//! +//! Memory analogue of [`crate::shard::disk_monitor`] (MA12): pauses writes +//! before the kernel OOM-killer can pick this process, by tracking ACTUAL +//! process RSS against the detected system/cgroup memory limit — NOT the +//! configured `--maxmemory`, which can be an unconfigured `0` (unlimited). +//! +//! # Design +//! +//! One `Arc` is created per server (not per shard). Shard 0's +//! event loop polls RSS every 5 seconds via `poll`. All other shards share +//! the same `Arc` and call `paused()` on the hot path. +//! +//! ## Hot-path cost +//! +//! `paused()` is a single `AtomicBool::load(Relaxed)`. No syscall, no heap +//! allocation, no lock. +//! +//! ## Hysteresis — direction INVERTED vs the disk monitor +//! +//! The disk monitor pauses when free space is LOW (bad) and resumes when it +//! rises. This monitor pauses when RSS is HIGH (bad) and resumes when it +//! FALLS. Concretely: +//! +//! - Writes pause once `rss * 100 / limit >= pause_pct`. +//! - Writes resume only once `rss * 100 / limit <= pause_pct - HYSTERESIS_PCT`. +//! +//! The gap prevents flapping around the threshold as allocator arenas and +//! GC-style reclamation shrink RSS incrementally. +//! +//! ## Fixed limit, unlike the disk monitor's per-poll total +//! +//! `disk_monitor::poll` re-queries both free AND total bytes every 5s because +//! a volume can be resized. The memory limit is different: detecting it can +//! shell out (`sysctl` on macOS) or read `/proc` + `/sys/fs/cgroup` (Linux), +//! and the caller-supplied context requires this NEVER happen on the hot +//! poll path. So `limit_bytes` is resolved ONCE at [`init_global`] time and +//! held fixed for the process lifetime; only `rss_bytes` is re-sampled. +//! +//! ## Read failures +//! +//! [`crate::admin::metrics_setup::get_rss_bytes`] returns `0` both on read +//! failure AND (theoretically) on platforms with no RSS probe. Since a live +//! server process never legitimately has 0 RSS, a `0` reading is always +//! treated as "read failed — retain previous state" (mirrors +//! `disk_monitor::poll`'s `None` arm). +//! +//! ## Accepted trade-off (same as MA12) +//! +//! Read-only commands are never blocked (the write gate already handles +//! this). `DEL` / `UNLINK` / `EXPIRE` / `FLUSHALL` are write-flagged and ARE +//! blocked while paused, exactly like the diskfull guard — no allowlist in +//! this module; that would need separate policy design. +//! +//! ## Test/diagnostic override: `MOON_MEM_LIMIT_BYTES` +//! +//! [`init_global`] honors `MOON_MEM_LIMIT_BYTES` (parsed as `u64`) in place +//! of [`crate::config::detect_memory_limit_bytes`] when set. This exists so +//! integration tests can deterministically engage the watchdog without +//! controlling real process RSS or faking `/proc`/`/sys/fs/cgroup` — the same +//! rationale as the `MOON_XSHARD_*` bench-diagnostic knobs (see CLAUDE.md). +//! Not a production tuning knob: operators size the real limit via cgroups / +//! container memory requests, not this env var. + +use std::sync::Arc; +use std::sync::OnceLock; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +/// Hysteresis gap: writes resume when `rss_pct <= pause_pct - HYSTERESIS_PCT`. +const HYSTERESIS_PCT: u8 = 5; + +/// Test/diagnostic override for the detected memory limit. See module docs. +const MEM_LIMIT_OVERRIDE_ENV: &str = "MOON_MEM_LIMIT_BYTES"; + +/// Proactive RSS memory watchdog. +/// +/// Create one instance per server, wrap in `Arc`, share across shards. +/// +/// ``` +/// # use moon::shard::mem_monitor::MemMonitor; +/// # use std::sync::Arc; +/// let monitor = Arc::new(MemMonitor::new(95, 16 * 1024 * 1024 * 1024)); +/// // Shard 0 timer: monitor.poll(); +/// // Hot path: if monitor.paused() { return error; } +/// ``` +pub struct MemMonitor { + /// Latest measured RSS in bytes. + rss_bytes: AtomicU64, + /// True when writes should be refused (rss% >= pause_pct). + paused: AtomicBool, + /// RSS percentage of `limit_bytes` at which writes pause (inclusive). + pause_pct: u8, + /// Detected system/cgroup memory limit in bytes. Fixed for the process + /// lifetime (resolved once by `init_global`). `0` means "undetectable" — + /// monitoring is permanently disabled regardless of `pause_pct`. + limit_bytes: u64, +} + +impl MemMonitor { + /// Create a monitor. + /// + /// * `pause_pct` — pause writes when `rss * 100 / limit_bytes >= pause_pct`. + /// * `limit_bytes` — detected system/cgroup memory limit; `0` disables + /// monitoring entirely (limit is undetectable). + /// + /// Starts unpaused with `rss_bytes = 0` (optimistic: no pause before the + /// first `poll` fires — mirrors `DiskMonitor::new`'s optimistic default, + /// just inverted: 0 bytes used is the "healthy" extreme for RSS). + pub fn new(pause_pct: u8, limit_bytes: u64) -> Self { + Self { + rss_bytes: AtomicU64::new(0), + paused: AtomicBool::new(false), + pause_pct, + limit_bytes, + } + } + + /// Latest measured RSS in bytes. + /// + /// Returns `0` until the first `poll` completes successfully. + #[inline] + pub fn rss_bytes(&self) -> u64 { + self.rss_bytes.load(Ordering::Relaxed) + } + + /// Returns `true` when writes should be refused. + /// + /// Extremely cheap: single `AtomicBool::load(Relaxed)`. + #[inline] + pub fn paused(&self) -> bool { + self.paused.load(Ordering::Relaxed) + } + + /// Current configured pause-percentage threshold. + #[inline] + pub fn pause_pct(&self) -> u8 { + self.pause_pct + } + + /// Detected memory limit in bytes (`0` = undetectable / disabled). + #[inline] + pub fn limit_bytes(&self) -> u64 { + self.limit_bytes + } + + /// Sample process RSS and update internal state. + /// + /// Called by shard 0's timer every 5 seconds. Safe to call from any + /// shard, but must not be called on every write (it is not free). + pub fn poll(&self) { + let rss = crate::admin::metrics_setup::get_rss_bytes(); + if rss == 0 { + // Read failure (or a platform with no RSS probe) — leave previous + // state unchanged to avoid spurious pauses due to transient + // errors. Mirrors disk_monitor::poll's `None` arm. + tracing::warn!("mem_monitor: get_rss_bytes returned 0; retaining previous pause state"); + return; + } + self.rss_bytes.store(rss, Ordering::Relaxed); + self.update_paused(rss); + // Wire P10 INFO metrics (mirrors disk_monitor's RECL_* wiring). + crate::command::info_reclamation::RECL_MEM_RSS_BYTES.store(rss, Ordering::Relaxed); + crate::command::info_reclamation::RECL_MEM_WATCHDOG_ACTIVE.store( + if self.paused.load(Ordering::Relaxed) { + 1 + } else { + 0 + }, + Ordering::Relaxed, + ); + } + + /// Inject a known RSS reading — used by tests to bypass the syscall. + /// + /// Not part of the public API surface; only `pub(crate)` for unit tests. + #[cfg(test)] + pub fn inject(&self, rss: u64) { + self.rss_bytes.store(rss, Ordering::Relaxed); + self.update_paused(rss); + } + + /// Core hysteresis state machine. Must only be called from `poll`/`inject`. + /// + /// Direction is INVERTED vs `DiskMonitor::update_paused`: high RSS is + /// bad, so the enter-pause comparison is `>=` (inclusive at the + /// threshold — unlike the disk monitor's strict `<`), and the resume + /// comparison is `<=` a LOWER threshold. + fn update_paused(&self, rss: u64) { + if self.limit_bytes == 0 || self.pause_pct == 0 { + // Undetectable limit or explicitly disabled — never pause. + return; + } + // Clamp before the u64->u8 cast: RSS can exceed limit_bytes (a + // stale/under-detected cgroup limit, or a transient spike right + // before the kernel OOM-killer would act), and an unclamped + // percentage > 255 would truncate-wrap through `as u8` into a + // small, misleadingly "healthy" value. + let rss_pct = (rss.saturating_mul(100) / self.limit_bytes).min(255) as u8; + let currently_paused = self.paused.load(Ordering::Relaxed); + + if !currently_paused && rss_pct >= self.pause_pct { + // Enter paused state + self.paused.store(true, Ordering::Release); + tracing::warn!( + rss_pct, + pause_pct = self.pause_pct, + rss_bytes = rss, + limit_bytes = self.limit_bytes, + "mem_monitor: write stall ENGAGED — RSS approaching memory limit", + ); + } else if currently_paused { + let resume_pct = self.pause_pct.saturating_sub(HYSTERESIS_PCT); + if rss_pct <= resume_pct { + self.paused.store(false, Ordering::Release); + tracing::info!( + rss_pct, + resume_pct, + "mem_monitor: write stall CLEARED — RSS pressure recovered", + ); + } + } + } +} + +// ── Process-global singleton ──────────────────────────────────────────────── +// +// One `Arc` per server process, initialised once at startup by +// `init_global`. All shards read `GLOBAL_MEM_MONITOR` on the hot path; shard 0 +// calls `poll_global` every 5 seconds from its timer (same tick as MA12). + +static GLOBAL_MEM_MONITOR: OnceLock> = OnceLock::new(); + +/// Initialise the process-global memory monitor. +/// +/// Must be called once at server startup (before any shard event loops +/// start). Calling it more than once is a no-op — the first call wins. +/// +/// Resolves `limit_bytes` from `MOON_MEM_LIMIT_BYTES` (test/diagnostic +/// override, see module docs) if set, else +/// [`crate::config::detect_memory_limit_bytes`] (ONE call — this may shell +/// out on macOS, so it must never run on the per-poll path). +/// +/// Runs one immediate `poll()` before publishing the monitor so the startup +/// recovery peak (AOF replay + segment load) is visible to the guard before +/// the first 5s timer tick. +/// +/// * `pause_pct` — from `ServerConfig::mem_full_pct`. `0` disables the guard. +pub fn init_global(pause_pct: u8) { + let limit_bytes = std::env::var(MEM_LIMIT_OVERRIDE_ENV) + .ok() + .and_then(|v| v.parse::().ok()) + .or_else(|| crate::config::detect_memory_limit_bytes().map(|v| v as u64)) + .unwrap_or(0); + let monitor = Arc::new(MemMonitor::new(pause_pct, limit_bytes)); + if pause_pct > 0 { + monitor.poll(); + } + // Ignore the error: if already set, the existing instance remains. + let _ = GLOBAL_MEM_MONITOR.set(monitor); +} + +/// Poll the global monitor (shard 0 calls this every 5 seconds). +/// +/// No-op if `init_global` has not been called or if `pause_pct == 0` +/// (monitoring disabled). +pub fn poll_global() { + if let Some(m) = GLOBAL_MEM_MONITOR.get() { + if m.pause_pct > 0 { + m.poll(); + } + } +} + +/// Returns `true` if writes should be refused due to memory pressure. +/// +/// **Hot-path function.** Single `AtomicBool::load(Relaxed)` when monitoring +/// is active; returns `false` immediately when monitoring is disabled or the +/// global has not been initialised. +#[inline] +pub fn is_write_paused() -> bool { + match GLOBAL_MEM_MONITOR.get() { + Some(m) => m.paused(), + None => false, + } +} + +/// Returns the latest measured RSS in bytes (for INFO output). +/// +/// Returns `0` when monitoring is disabled or not yet initialised. +#[inline] +pub fn global_rss_bytes() -> u64 { + match GLOBAL_MEM_MONITOR.get() { + Some(m) => m.rss_bytes(), + None => 0, + } +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + /// Helper that creates a monitor and injects an RSS reading via the test + /// backdoor, then returns the monitor so callers can assert state. + fn monitor_with(pause_pct: u8, rss: u64, limit: u64) -> MemMonitor { + let m = MemMonitor::new(pause_pct, limit); + m.inject(rss); + m + } + + // ── Pause threshold (INVERTED vs disk: high RSS is bad) ──────────────── + + #[test] + fn test_paused_when_above_threshold() { + // 60% used, threshold 50% → should pause + let m = monitor_with(50, 60, 100); + assert!(m.paused(), "should be paused when rss% >= pause_pct"); + } + + #[test] + fn test_not_paused_when_below_threshold() { + // 40% used, threshold 50% → should NOT pause + let m = monitor_with(50, 40, 100); + assert!(!m.paused(), "should not be paused when rss% < pause_pct"); + } + + #[test] + fn test_paused_at_exact_threshold() { + // Exactly 50% used, threshold 50% → INVERTED vs disk: >= is inclusive, + // so this SHOULD pause (disk's exact-threshold case does NOT pause). + let m = monitor_with(50, 50, 100); + assert!( + m.paused(), + "rss% == pause_pct MUST pause (>= is inclusive, inverted vs disk's < )" + ); + } + + #[test] + fn test_rss_bytes_accessor() { + let m = monitor_with(50, 42_000_000, 1_000_000_000); + assert_eq!(m.rss_bytes(), 42_000_000); + } + + // ── Hysteresis ───────────────────────────────────────────────────────── + + #[test] + fn test_hysteresis_no_resume_at_pause_threshold() { + let m = MemMonitor::new(50, 100); + // Enter paused state: 60% used + m.inject(60); + assert!(m.paused(), "should be paused at 60%"); + + // Drop to exactly pause_pct (50%) — NOT below pause_pct - HYSTERESIS (45%) + m.inject(50); + assert!( + m.paused(), + "should remain paused at exactly pause_pct (hysteresis prevents premature resume)" + ); + } + + #[test] + fn test_hysteresis_resume_below_resume_threshold() { + let m = MemMonitor::new(50, 100); + // Enter paused state + m.inject(60); + assert!(m.paused(), "should be paused at 60%"); + + // Drop to pause_pct - HYSTERESIS (45%) — should resume + let resume_pct = 50u64 - HYSTERESIS_PCT as u64; + m.inject(resume_pct); + assert!( + !m.paused(), + "should resume when rss% <= pause_pct - HYSTERESIS_PCT" + ); + } + + #[test] + fn test_hysteresis_flap_prevention() { + let m = MemMonitor::new(50, 100); + + // Enter paused + m.inject(70); + assert!(m.paused()); + + // Yo-yo between 46% and 49% — stays paused because 46 > resume(45) + m.inject(46); + assert!(m.paused(), "46% > 45% resume threshold, should stay paused"); + m.inject(49); + assert!(m.paused()); + m.inject(46); + assert!(m.paused(), "still paused — never crossed 45%"); + + // Cross 45% — should resume + m.inject(45); + assert!(!m.paused(), "45% <= 45% resume threshold, should clear"); + } + + #[test] + fn test_no_flap_after_resume() { + let m = MemMonitor::new(50, 100); + + // Normal operation: rss is healthy + m.inject(10); + assert!(!m.paused()); + + // Rise above threshold + m.inject(70); + assert!(m.paused()); + + // Recover past hysteresis + m.inject(30); + assert!(!m.paused()); + + // Another rise to pause_pct — should pause again + m.inject(55); + assert!( + m.paused(), + "re-enter paused on another rise above threshold" + ); + } + + // ── Edge cases ───────────────────────────────────────────────────────── + + #[test] + fn test_zero_limit_disabled_does_not_panic() { + let m = MemMonitor::new(50, 0); + // inject a huge rss against a zero limit — should not panic (no + // division-by-zero), and monitoring must stay disabled. + m.inject(u64::MAX); + assert!( + !m.paused(), + "limit_bytes == 0 (undetectable) must permanently disable the guard" + ); + } + + #[test] + fn test_pause_pct_zero_disabled() { + let m = MemMonitor::new(0, 100); + // Even at 100% RSS usage, pause_pct == 0 means "disabled". + m.inject(100); + assert!(!m.paused(), "pause_pct == 0 must disable the guard"); + } + + #[test] + fn test_rss_exceeding_limit_does_not_wrap_u8() { + // rss is 10x the limit → naive `as u8` on an unclamped percentage + // (1000) would truncate-wrap to 1000 % 256 = 232, which happens to + // still be >= most thresholds, but at other multiples the wrap can + // land BELOW pause_pct and falsely clear the guard. Pin the case + // that would silently regress if the `.min(255)` clamp is removed. + let m = MemMonitor::new(50, 100); + m.inject(2560); // 2560*100/100 = 2560 -> wraps to 0 without clamping + assert!( + m.paused(), + "rss far exceeding limit_bytes must still register as paused, \ + not wrap through u8 truncation into a false 'healthy' reading" + ); + } + + #[test] + fn test_large_memory_host() { + // 64 GiB limit, 40 GiB RSS = 62.5% → above a 50% threshold. + let limit: u64 = 64 * 1024 * 1024 * 1024; + let rss: u64 = 40 * 1024 * 1024 * 1024; + let m = monitor_with(50, rss, limit); + assert!( + m.paused(), + "62.5% used on a large host should trigger pause" + ); + } + + #[test] + fn test_initial_state_optimistic() { + // Before first poll, monitor should be unpaused (optimistic default). + let m = MemMonitor::new(50, 100); + assert!(!m.paused(), "initial state must be unpaused"); + assert_eq!(m.rss_bytes(), 0, "initial rss_bytes must be 0"); + } + + /// Smoke test: poll() on the real process must not panic and must update + /// rss_bytes from the 0 sentinel to some positive value (a live server + /// process never legitimately has 0 RSS). + #[test] + fn test_poll_real_process_smoke() { + let m = MemMonitor::new(50, u64::MAX); + m.poll(); + assert!( + m.rss_bytes() > 0, + "real get_rss_bytes() must update rss_bytes on a live process" + ); + } + + #[test] + fn test_arc_send_sync() { + // MemMonitor must be Send + Sync for cross-shard sharing via Arc. + fn assert_send_sync() {} + assert_send_sync::(); + } +} diff --git a/src/shard/mod.rs b/src/shard/mod.rs index 6904cda9a..0fca0ce6f 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -8,6 +8,9 @@ pub mod dispatch; pub mod event_loop; /// MA5: maintenance-window scheduler (cron-style budget multipliers). pub mod maintenance_schedule; +/// Wave 3: proactive RSS memory watchdog ("mem-full guard") — analogue of +/// `disk_monitor` (MA12) for process RSS vs the detected system/cgroup limit. +pub mod mem_monitor; pub mod mesh; /// C2 (shardslice-migration Wave A1): owner-side MQ.* execution on the shard thread. pub(crate) mod mq_exec; From a98eaa8cf05b4d258a5e9420850f6e0ae4b11521 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 16:11:22 +0700 Subject: [PATCH 2/4] feat(server): wire mem_monitor into write-stall dispatch and shard poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires the mem_monitor core (previous commit) into the live write path: - segment_stall::is_any_write_stall_active() OR-merges a third source, mem_monitor::is_write_paused(), alongside the existing MA12 disk-pressure and MA1 segment-backlog bits. Hot-path cost: one additional AtomicBool::load(Relaxed) inside the already write-gated branch — no allocation, no lock. - Both dispatch paths' try_enforce_disk_full (handler_monoio and handler_sharded) gain a MOONERR memfull branch in the message-selection chain: diskfull first, then memfull, then the segment-backlog busy fallback — mirrors the existing diskfull/busy distinction verbatim. - event_loop.rs: mem_monitor::poll_global() runs on shard 0's existing 5s disk_monitor timer tick, at BOTH poll sites (tokio select! arm and the monoio tick-counter branch) — no new timer, reuses MA12's cadence. - main.rs: mem_monitor::init_global(config.mem_full_pct) called right after disk_monitor::init_global, the single call site for both guards. Runs one immediate poll inside init_global (see previous commit) so the AOF-replay /segment-load startup recovery peak is visible before the first 5s tick. author: Tin Dang --- src/main.rs | 8 ++++++++ src/server/conn/handler_monoio/dispatch.rs | 7 +++++-- src/server/conn/handler_sharded/dispatch.rs | 7 +++++-- src/shard/event_loop.rs | 4 ++++ src/shard/segment_stall.rs | 19 +++++++++++++------ 5 files changed, 35 insertions(+), 10 deletions(-) diff --git a/src/main.rs b/src/main.rs index e4d9aa2b4..a87333a43 100644 --- a/src/main.rs +++ b/src/main.rs @@ -889,6 +889,14 @@ fn main() -> anyhow::Result<()> { moon::shard::disk_monitor::init_global(config.disk_free_min_pct, monitor_path); } + // Wave 3: Initialise proactive RSS memory watchdog ("mem-full guard"). + // Fires on ACTUAL RSS vs the detected system/cgroup limit, not on + // --maxmemory (which can be an unconfigured 0). When mem_full_pct == 0, + // the monitor is inactive (poll_global is a no-op, is_write_paused always + // false). Runs one immediate poll so the startup recovery peak (AOF + // replay + segment load) is visible before the first 5s timer tick. + moon::shard::mem_monitor::init_global(config.mem_full_pct); + // Collect all notifiers before spawning shard threads let all_notifiers = mesh.all_notifiers(); diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 336a39682..c833330a7 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -687,21 +687,24 @@ pub(super) fn try_enforce_readonly( false } -/// MA12 + MA1: Refuse write commands when any write stall is active. +/// MA12 + MA1 + Wave 3: Refuse write commands when any write stall is active. /// /// Returns `true` if the command was blocked (caller should `continue`). /// /// Stall sources (OR-merged): /// - MA12 disk-pressure monitor (`is_write_paused`) — set every 5s. /// - MA1 segment-backlog stall (`is_segment_stall_active`) — set every 1s. +/// - Wave 3 RSS memory watchdog (`mem_monitor::is_write_paused`) — set every 5s. /// -/// Hot path: two `Atomic::load(Relaxed)` — no allocation, no lock. +/// Hot path: three `Atomic::load(Relaxed)` — no allocation, no lock. /// Read-only commands pass through unaffected. Background compaction is exempt. #[inline] pub(super) fn try_enforce_disk_full(cmd: &[u8], responses: &mut Vec) -> bool { if metadata::is_write(cmd) && crate::shard::segment_stall::is_any_write_stall_active() { let msg: &'static [u8] = if crate::shard::disk_monitor::is_write_paused() { b"MOONERR diskfull: writes paused until free space recovers" + } else if crate::shard::mem_monitor::is_write_paused() { + b"MOONERR memfull: writes paused until memory pressure recovers" } else { b"MOONERR busy: compaction backlog; too many unflushed immutable segments" }; diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 6207cd467..73a2bffab 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -580,15 +580,16 @@ pub(super) fn try_enforce_readonly( false } -/// MA12 + MA1: Refuse write commands when any write stall is active. +/// MA12 + MA1 + Wave 3: Refuse write commands when any write stall is active. /// /// Returns `true` if the command was blocked (caller should `continue`). /// /// Stall sources (OR-merged): /// - MA12 disk-pressure monitor (`is_write_paused`) — set every 5s. /// - MA1 segment-backlog stall (`is_segment_stall_active`) — set every 1s. +/// - Wave 3 RSS memory watchdog (`mem_monitor::is_write_paused`) — set every 5s. /// -/// Hot path: two `Atomic::load(Relaxed)` — no allocation, no lock. +/// Hot path: three `Atomic::load(Relaxed)` — no allocation, no lock. /// Read-only commands pass through unaffected; only writes are stalled. /// Background compaction (FT.COMPACT, GRAPH.COMPACT) is exempt. #[inline] @@ -597,6 +598,8 @@ pub(super) fn try_enforce_disk_full(cmd: &[u8], responses: &mut Vec) -> b // Distinguish the stall source for operator clarity. let msg: &'static [u8] = if crate::shard::disk_monitor::is_write_paused() { b"MOONERR diskfull: writes paused until free space recovers" + } else if crate::shard::mem_monitor::is_write_paused() { + b"MOONERR memfull: writes paused until memory pressure recovers" } else { b"MOONERR busy: compaction backlog; too many unflushed immutable segments" }; diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 4734c2a33..1fff171c0 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1639,6 +1639,8 @@ impl super::Shard { _ = disk_monitor_interval.0.tick() => { if shard_id == 0 { crate::shard::disk_monitor::poll_global(); + // Wave 3: RSS memory watchdog poll (same 5s tick). + crate::shard::mem_monitor::poll_global(); } } // P4: Autovacuum daemon tick (default 30s interval). @@ -2237,6 +2239,8 @@ impl super::Shard { // MA12: Disk free-space poll (every 5000 ticks = 5s, shard 0 only). if shard_id == 0 && monoio_tick_counter % 5000 == 0 { crate::shard::disk_monitor::poll_global(); + // Wave 3: RSS memory watchdog poll (same 5s tick). + crate::shard::mem_monitor::poll_global(); } // P4: Autovacuum daemon tick (every autovacuum_interval_secs * 1000 ticks). if monoio_tick_counter % (autovacuum_interval_secs * 1000) == 0 diff --git a/src/shard/segment_stall.rs b/src/shard/segment_stall.rs index f6b60b36a..76460bfa4 100644 --- a/src/shard/segment_stall.rs +++ b/src/shard/segment_stall.rs @@ -10,9 +10,11 @@ //! //! - **MA12 (disk monitor):** sets `RECL_WRITE_STALL_ACTIVE` on disk-pressure events. //! - **MA1 (this module):** sets `RECL_SEGMENT_STALL_ACTIVE` on segment-backlog events. -//! - **INFO `write_stall_active`:** emits `true` if EITHER bit is non-zero (OR of both). +//! - **Wave 3 (mem monitor):** sets `RECL_MEM_WATCHDOG_ACTIVE` on RSS-pressure events. +//! - **INFO `write_stall_active`:** emits `true` if ANY bit is non-zero (OR of all three). //! - **Dispatch `try_enforce_write_stall`:** calls `is_any_write_stall_active()` which -//! ORs `is_write_paused()` (MA12) with `is_segment_stall_active()` (MA1). +//! ORs `is_write_paused()` (MA12) with `is_segment_stall_active()` (MA1) with +//! `mem_monitor::is_write_paused()` (Wave 3). //! //! ## Hot-path cost //! @@ -39,15 +41,20 @@ pub fn is_segment_stall_active() -> bool { RECL_SEGMENT_STALL_ACTIVE.load(Ordering::Relaxed) != 0 } -/// Returns `true` if ANY write stall is active (disk-pressure OR segment-backlog). +/// Returns `true` if ANY write stall is active (disk-pressure OR segment-backlog +/// OR RSS memory pressure). /// -/// This is the unified check used by both dispatch paths. OR-merges MA12 and MA1: +/// This is the unified check used by both dispatch paths. OR-merges MA12, MA1, +/// and Wave 3: /// -/// - `is_write_paused()` — MA12 disk-free monitor (AtomicBool, set every 5s). +/// - `disk_monitor::is_write_paused()` — MA12 disk-free monitor (AtomicBool, set every 5s). /// - `is_segment_stall_active()` — MA1 segment backlog (AtomicU64, set every 1s). +/// - `mem_monitor::is_write_paused()` — Wave 3 RSS watchdog (AtomicBool, set every 5s). #[inline] pub fn is_any_write_stall_active() -> bool { - crate::shard::disk_monitor::is_write_paused() || is_segment_stall_active() + crate::shard::disk_monitor::is_write_paused() + || is_segment_stall_active() + || crate::shard::mem_monitor::is_write_paused() } /// Update the segment-stall atomic based on the current immutable segment count. From cb1f6424e2024e66f1bf273ed230085f92c23a1b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 16:11:42 +0700 Subject: [PATCH 3/4] test(shard): mem_watchdog integration suite + ServerConfig literal fixups MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds tests/mem_watchdog.rs, the wire-level counterpart to mem_monitor's unit tests: proves the --mem-full-pct CLI flag, init_global's immediate poll, and the dispatch-level MOONERR memfull message path against a real spawned server (pattern: tests/oom_bypass_closure.rs). Real process RSS and the detected memory limit aren't controllable from an external test process, so the suite drives the guard via the MOON_MEM_LIMIT_BYTES test override documented in mem_monitor's module header: - Case A: MOON_MEM_LIMIT_BYTES=1MB + --mem-full-pct 50 -> the FIRST SET is already refused with MOONERR memfull (init_global's immediate poll, no need to wait for the 5s tick). - Case B: GET is never blocked while the guard is engaged (read-only commands are exempt via the existing is_write gate). - Case C: control — no env override, SET succeeds normally (negative control against spurious engagement). - Case D: --mem-full-pct 0 disables the guard even with a tiny forced limit that would otherwise engage it at any nonzero threshold. RED confirmed by running this suite against a binary built BEFORE the mem_monitor implementation (git stash of all impl commits, rebuild, all 4 cases fail with "server never accepted" because --mem-full-pct doesn't parse yet); GREEN confirmed by unstashing and rebuilding — all 4 pass on both the default (monoio) and runtime-tokio,jemalloc feature sets. Also fixes 3 pre-existing test files that construct ServerConfig via an EXHAUSTIVE struct literal (no ..Default::default() spread), which the new mem_full_pct field broke at compile time: tests/mq_integration.rs, tests/txn_kv_wiring.rs, tests/workspace_integration.rs (2 occurrences). Each gets mem_full_pct: 0 (disabled) next to its existing disk_free_min_pct: 5 line — these tests don't exercise the RSS watchdog and 0 preserves their prior (guard-absent) behavior exactly. author: Tin Dang --- tests/mem_watchdog.rs | 371 +++++++++++++++++++++++++++++++++ tests/mq_integration.rs | 1 + tests/txn_kv_wiring.rs | 1 + tests/workspace_integration.rs | 2 + 4 files changed, 375 insertions(+) create mode 100644 tests/mem_watchdog.rs diff --git a/tests/mem_watchdog.rs b/tests/mem_watchdog.rs new file mode 100644 index 000000000..20b1dc138 --- /dev/null +++ b/tests/mem_watchdog.rs @@ -0,0 +1,371 @@ +//! Wave 3 — proactive RSS memory watchdog ("mem-full guard") integration test. +//! +//! `mem_monitor` (`src/shard/mem_monitor.rs`) is the memory analogue of the +//! existing `disk_monitor` (MA12) diskfull guard: it pauses writes once +//! process RSS crosses `--mem-full-pct` of the detected system/cgroup memory +//! limit, instead of waiting for the kernel OOM-killer to pick this process. +//! +//! Real process RSS cannot be steered from an external test process, and the +//! detected limit (`detect_memory_limit_bytes`) isn't controllable either — +//! so this suite drives the guard via the `MOON_MEM_LIMIT_BYTES` test/ +//! diagnostic env override documented in `mem_monitor`'s module header: set +//! it far below the server's real (unavoidable, always > 0) startup RSS, and +//! the very first `init_global` poll engages the pause deterministically. +//! +//! Wire-level on purpose: unit tests in `mem_monitor.rs` already cover the +//! hysteresis state machine in isolation. This suite instead proves the +//! dispatch-level wiring — CLI flag parsing, `init_global`'s immediate poll, +//! and the `MOONERR memfull` error path — the way `tests/oom_bypass_closure.rs` +//! proves the `--maxmemory` eviction gate's wiring. +//! +//! Run alone with: +//! MOON_BIN=$PWD/target/debug/moon cargo test --test mem_watchdog + +#![allow(clippy::unwrap_used)] + +use std::io::{BufReader, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::process::{Child, Command}; +use std::time::{Duration, Instant}; + +// --------------------------------------------------------------------------- +// Binary resolution + server spawn (pattern: tests/oom_bypass_closure.rs) +// --------------------------------------------------------------------------- + +fn find_moon_binary() -> std::path::PathBuf { + if let Ok(bin) = std::env::var("MOON_BIN") { + let p = std::path::PathBuf::from(bin); + if p.exists() { + return p; + } + } + let manifest = env!("CARGO_MANIFEST_DIR"); + let release = std::path::PathBuf::from(format!("{manifest}/target/release/moon")); + if release.exists() { + return release; + } + let debug = std::path::PathBuf::from(format!("{manifest}/target/debug/moon")); + if debug.exists() { + return debug; + } + panic!("No moon binary found. Build first or set MOON_BIN=/path/to/moon."); +} + +/// Ports below 20000 collide with other services in CI/dev; pick a free one +/// above that floor instead of a fixed low port. +fn free_port() -> u16 { + loop { + let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind :0"); + let p = l.local_addr().expect("local_addr").port(); + drop(l); + if p >= 20000 { + return p; + } + } +} + +/// `tempfile::tempdir()` defaults to `$TMPDIR`, which on macOS lives on the +/// root volume group — observed near Moon's 5%-free diskfull write-pause +/// guard (`MOONERR diskfull`) in dev environments. Root the test's scratch +/// dirs under the repo's own volume instead (see +/// gotcha_vm_diskfull_shared_volume in project memory). +fn test_tmpdir() -> tempfile::TempDir { + let base = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("target/mem-watchdog-tmp"); + std::fs::create_dir_all(&base).expect("create mem-watchdog-tmp base dir"); + tempfile::Builder::new() + .prefix("mem-watchdog-") + .tempdir_in(&base) + .expect("tempdir_in target/mem-watchdog-tmp") +} + +struct ServerGuard(Child); + +impl Drop for ServerGuard { + fn drop(&mut self) { + // kill() sends SIGKILL on all platforms via std::process::Child; + // belt-and-suspenders backstop against a leaked busy-poller (see + // gotcha_leaked_moon_busypoller_contaminates_xshard in project memory). + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Spawn moon with `--mem-full-pct ` and, optionally, +/// `MOON_MEM_LIMIT_BYTES=` to deterministically engage (or, when +/// `None`, leave untouched — falling back to real limit detection) the RSS +/// watchdog. +fn spawn_moon_mem( + port: u16, + dir: &std::path::Path, + mem_full_pct: u8, + limit_bytes_override: Option, +) -> ServerGuard { + let mut cmd = Command::new(find_moon_binary()); + cmd.args([ + "--port", + &port.to_string(), + "--dir", + &dir.to_string_lossy(), + "--shards", + "1", + "--appendonly", + "no", + "--mem-full-pct", + &mem_full_pct.to_string(), + ]); + if let Some(limit) = limit_bytes_override { + cmd.env("MOON_MEM_LIMIT_BYTES", limit.to_string()); + } else { + // Ensure no ambient override leaks in from the test runner's shell. + cmd.env_remove("MOON_MEM_LIMIT_BYTES"); + } + let child = 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"); + ServerGuard(child) +} + +// --------------------------------------------------------------------------- +// Minimal RESP client (binary-safe args, full-frame parser) +// --------------------------------------------------------------------------- + +#[derive(Debug, PartialEq)] +enum V { + Simple(String), + Err(String), + Bulk(Vec), + Null, +} + +impl V { + fn is_memfull_error(&self) -> bool { + matches!(self, V::Err(msg) if msg.to_uppercase().contains("MEMFULL")) + } +} + +struct Client { + reader: BufReader, + writer: TcpStream, +} + +impl Client { + fn connect(port: u16) -> Self { + let addr = format!("127.0.0.1:{port}") + .to_socket_addrs() + .unwrap() + .next() + .unwrap(); + let start = Instant::now(); + let stream = loop { + match TcpStream::connect_timeout(&addr, Duration::from_millis(200)) { + Ok(s) => break s, + Err(_) if start.elapsed() < Duration::from_secs(30) => { + std::thread::sleep(Duration::from_millis(50)); + } + Err(e) => panic!("server never accepted on port {port}: {e}"), + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(30))) + .unwrap(); + let writer = stream.try_clone().unwrap(); + Client { + reader: BufReader::new(stream), + writer, + } + } + + fn encode(args: &[&[u8]]) -> Vec { + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n", a.len()).as_bytes()); + out.extend_from_slice(a); + out.extend_from_slice(b"\r\n"); + } + out + } + + fn read_line(&mut self) -> String { + let mut line = Vec::new(); + let mut b = [0u8; 1]; + loop { + self.reader.read_exact(&mut b).expect("read byte"); + if b[0] == b'\n' { + break; + } + if b[0] != b'\r' { + line.push(b[0]); + } + } + String::from_utf8_lossy(&line).into_owned() + } + + fn parse(&mut self) -> V { + let line = self.read_line(); + let (t, rest) = line.split_at(1); + match t { + "+" => V::Simple(rest.to_string()), + "-" => V::Err(rest.to_string()), + ":" => { + let _n: i64 = rest.parse().expect("int"); + V::Simple(rest.to_string()) + } + "$" => { + let n: i64 = rest.parse().expect("bulk len"); + if n < 0 { + return V::Null; + } + let mut buf = vec![0u8; n as usize + 2]; + self.reader.read_exact(&mut buf).expect("bulk body"); + buf.truncate(n as usize); + V::Bulk(buf) + } + "*" => { + let n: i64 = rest.parse().expect("arr len"); + if n < 0 { + return V::Null; + } + for _ in 0..n { + self.parse(); + } + V::Null + } + other => panic!("unexpected RESP type {other:?} (line {line:?})"), + } + } + + fn cmd(&mut self, args: &[&[u8]]) -> V { + self.writer.write_all(&Self::encode(args)).expect("send"); + self.parse() + } + + /// Fallible PING for the readiness probe: a connection accepted while the + /// server is still bringing up its per-shard SO_REUSEPORT listeners can be + /// RESET mid-read — that must retry with a fresh connection, not panic. + fn try_ping(&mut self) -> std::io::Result { + self.writer.write_all(b"*1\r\n$4\r\nPING\r\n")?; + let mut buf = [0u8; 7]; + self.reader.read_exact(&mut buf)?; + Ok(&buf == b"+PONG\r\n") + } +} + +fn wait_ready(port: u16) -> Client { + let start = Instant::now(); + loop { + let mut c = Client::connect(port); + // Any I/O error (or a non-PONG answer, which would desync the framing) + // drops this connection and probes again on a new one. + if let Ok(true) = c.try_ping() { + return c; + } + assert!( + start.elapsed() < Duration::from_secs(30), + "server never answered PING on port {port}" + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +// --------------------------------------------------------------------------- +// Case A: MOON_MEM_LIMIT_BYTES set far below real startup RSS + a reachable +// --mem-full-pct → the FIRST write must already be refused with +// `MOONERR memfull`, proving `init_global`'s immediate poll (no need to wait +// for the 5s timer tick) and the dispatch-level message-selection wiring. +// --------------------------------------------------------------------------- + +#[test] +fn test_case_a_tiny_limit_engages_memfull_on_first_write() { + let dir = test_tmpdir(); + let port = free_port(); + // 1 MB "limit" is far below any real moon process's startup RSS — the + // watchdog's very first poll (inside init_global) must already compute + // rss% far past 50%. + const TINY_LIMIT_BYTES: u64 = 1_000_000; + let _guard = spawn_moon_mem(port, dir.path(), 50, Some(TINY_LIMIT_BYTES)); + let mut c = wait_ready(port); + + let r = c.cmd(&[b"SET", b"k1", b"v1"]); + assert!( + r.is_memfull_error(), + "expected MOONERR memfull on the first write with a 1MB fake limit \ + at 50% threshold, got {r:?}" + ); +} + +// --------------------------------------------------------------------------- +// Case B: read-only commands must NOT be blocked while the watchdog is +// engaged — the write gate (`metadata::is_write`) already exempts reads; +// this locks that invariant against a future over-broad gate. +// --------------------------------------------------------------------------- + +#[test] +fn test_case_b_get_not_blocked_while_memfull_engaged() { + let dir = test_tmpdir(); + let port = free_port(); + const TINY_LIMIT_BYTES: u64 = 1_000_000; + let _guard = spawn_moon_mem(port, dir.path(), 50, Some(TINY_LIMIT_BYTES)); + let mut c = wait_ready(port); + + // Confirm the guard is actually engaged (setup assertion). + let set_reply = c.cmd(&[b"SET", b"k2", b"v2"]); + assert!( + set_reply.is_memfull_error(), + "setup: SET should be blocked by the watchdog, got {set_reply:?}" + ); + + // GET must still work — read-only commands are exempt. + let get_reply = c.cmd(&[b"GET", b"k2"]); + assert_eq!( + get_reply, + V::Null, + "GET must not be blocked by the memfull guard (key was never written \ + due to the SET block above, so Null is the correct, non-error reply)" + ); +} + +// --------------------------------------------------------------------------- +// Case C: control — WITHOUT the env override (real limit detection, which +// on the current platform may return `None`/0 = undetectable, or a real +// large host limit either way) and a low `--mem-full-pct` floor is not +// artificially forced, SET must succeed normally. This is the negative +// control proving the guard doesn't fire spuriously on an ordinary server. +// --------------------------------------------------------------------------- + +#[test] +fn test_case_c_control_no_override_set_succeeds() { + let dir = test_tmpdir(); + let port = free_port(); + let _guard = spawn_moon_mem(port, dir.path(), 95, None); + let mut c = wait_ready(port); + + let r = c.cmd(&[b"SET", b"k3", b"v3"]); + assert_eq!( + r, + V::Simple("OK".into()), + "control: SET should succeed without a forced tiny memory limit, got {r:?}" + ); +} + +// --------------------------------------------------------------------------- +// Case D: `--mem-full-pct 0` disables the guard even with a tiny forced +// limit that would otherwise engage it at any nonzero threshold. +// --------------------------------------------------------------------------- + +#[test] +fn test_case_d_mem_full_pct_zero_disables_guard() { + let dir = test_tmpdir(); + let port = free_port(); + const TINY_LIMIT_BYTES: u64 = 1_000_000; + let _guard = spawn_moon_mem(port, dir.path(), 0, Some(TINY_LIMIT_BYTES)); + let mut c = wait_ready(port); + + let r = c.cmd(&[b"SET", b"k4", b"v4"]); + assert_eq!( + r, + V::Simple("OK".into()), + "--mem-full-pct 0 must disable the guard even with a tiny forced limit, got {r:?}" + ); +} diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs index 1f4fa9d38..9dac2eaad 100644 --- a/tests/mq_integration.rs +++ b/tests/mq_integration.rs @@ -107,6 +107,7 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) { manifest_tombstone_retain_epochs: 2, manifest_tombstone_retain_secs: 300, disk_free_min_pct: 5, + mem_full_pct: 0, mvcc_committed_prune_margin: 1000, max_unflushed_immutable_segments: 20, mvcc_old_snapshot_threshold_secs: 600, diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs index 35215dd88..dec81ea28 100644 --- a/tests/txn_kv_wiring.rs +++ b/tests/txn_kv_wiring.rs @@ -112,6 +112,7 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can manifest_tombstone_retain_epochs: 2, manifest_tombstone_retain_secs: 300, disk_free_min_pct: 5, + mem_full_pct: 0, mvcc_committed_prune_margin: 1000, max_unflushed_immutable_segments: 20, mvcc_old_snapshot_threshold_secs: 600, diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs index adabe2ec9..3817514a6 100644 --- a/tests/workspace_integration.rs +++ b/tests/workspace_integration.rs @@ -100,6 +100,7 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) { manifest_tombstone_retain_epochs: 2, manifest_tombstone_retain_secs: 300, disk_free_min_pct: 5, + mem_full_pct: 0, mvcc_committed_prune_margin: 1000, max_unflushed_immutable_segments: 20, mvcc_old_snapshot_threshold_secs: 600, @@ -329,6 +330,7 @@ async fn start_workspace_server_with_auth( manifest_tombstone_retain_epochs: 2, manifest_tombstone_retain_secs: 300, disk_free_min_pct: 5, + mem_full_pct: 0, mvcc_committed_prune_margin: 1000, max_unflushed_immutable_segments: 20, mvcc_old_snapshot_threshold_secs: 600, From 2e881571097b7a3afba9fc0d62fd4dea7ccb9b80 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 6 Jul 2026 16:11:53 +0700 Subject: [PATCH 4/4] docs(changelog): document proactive RSS watchdog (mem-full guard) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the [Unreleased] entry for the mem_monitor RSS watchdog (wave 3 of the RSS/CPU/OOM train) — summarizes the guard's semantics, the new --mem-full-pct flag, MOONERR memfull, and the new INFO fields, matching the format of the existing diskfull/OOM-bypass entries in this section. author: Tin Dang --- CHANGELOG.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index db7b97d10..d443696ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — proactive RSS watchdog pauses writes before kernel OOM (PR #TBD) + +- **New `mem_monitor` guard** (`src/shard/mem_monitor.rs`), the memory + analogue of the existing diskfull guard (MA12): pauses writes once process + RSS crosses `--mem-full-pct` (default 95, 0 = disabled) of the DETECTED + system/cgroup memory limit, not the configured `--maxmemory` (which can be + an unconfigured `0`/unlimited). Mirrors `disk_monitor`'s hysteresis state + machine, inverted for direction (high RSS is bad): pauses at + `rss% >= mem_full_pct`, resumes only at `rss% <= mem_full_pct - 5`. + `MOONERR memfull: writes paused until memory pressure recovers` joins the + existing `MOONERR diskfull` / `MOONERR busy` message-selection chain in + both dispatch paths (order: diskfull, memfull, segment-backlog busy); the + hot-path cost is one additional `AtomicBool::load(Relaxed)` inside the + already write-gated branch. Polled on shard 0's existing 5s disk-monitor + timer tick (both event-loop poll sites), plus one immediate poll at + startup (`init_global`) so the AOF-replay/segment-load recovery peak is + visible before the first tick. New INFO fields + `reclamation_mem_rss_bytes` / `reclamation_mem_watchdog_active` + (OR'd into the existing `reclamation_write_stall_active`). Same accepted + trade-off as the diskfull guard: `DEL`/`UNLINK`/`EXPIRE`/`FLUSHALL` are + write-flagged and blocked too while paused — no allowlist. + ### Fixed — zombie/CPU hardening wave 1 (PR #TBD) - **Busy-poll spin idle-disengages** (`--io-busy-poll-us` / vendored monoio