feat(shard): proactive RSS watchdog pauses writes before kernel OOM (RSS/CPU wave 3)

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

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() {

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<usize> {
 /// (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<usize> {
+pub(crate) fn detect_memory_limit_bytes() -> Option<usize> {
     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<usize> {
 /// 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<usize> {
+pub(crate) fn detect_memory_limit_bytes() -> Option<usize> {
     let out = std::process::Command::new("sysctl")
         .args(["-n", "hw.memsize"])
         .output()
@@ -1044,7 +1063,7 @@ fn detect_memory_limit_bytes() -> Option<usize> {
 /// 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<usize> {
+pub(crate) fn detect_memory_limit_bytes() -> Option<usize> {
     None
 }
 

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();
 

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<Frame>) -> 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"
         };

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<Frame>) -> 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"
         };

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