fix(shard): OOM shield — macOS maxmemory guardrail + eviction bypass closure (RSS/CPU wave 2)

CHANGELOG.md

@@ -29,6 +29,137 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   fresh connection on mid-startup connection resets (CI flake: per-shard
   SO_REUSEPORT listeners accept-then-reset during init).
 
+### Fixed — OOM eviction bypass closure (PR #TBD)
+
+- **Cross-shard SPSC write legs now enforce `--maxmemory`.** `Execute`,
+  `MultiExecute`, `PipelineBatch` and their `*Slotted` variants
+  (`src/shard/spsc_handler.rs`) executed writes against the TARGET shard's
+  `Database` directly, bypassing the connection handlers' eviction gate
+  entirely — a scatter-gather write (e.g. a pipeline of individually-routed
+  `SET`s hashing to a remote shard) could grow that shard's memory past
+  `maxmemory` without limit. A new `spsc_eviction_gate` helper mirrors
+  `run_write_eviction_gate` (`handler_monoio/mod.rs`) exactly — same elastic
+  per-shard budget (GAP-1), same spill-vs-plain branching — threaded through
+  `drain_spsc_shared`/`handle_shard_message_shared` and gated by a
+  once-per-drain-cycle `evict_active` snapshot (perf parity with
+  `batch_eviction_active`: zero extra lock acquires when `maxmemory` is
+  unset). **Perf follow-up (same PR):** that snapshot, and the Lua bridge's
+  gate below, each used to answer "is `maxmemory` nonzero?" with either a
+  `runtime_config.read()` per drain cycle or a generation/`Cell` snapshot.
+  Both now read a single process-global `AtomicU64`
+  (`storage::eviction::{publish_maxmemory, maxmemory_is_set}`), published at
+  every production write site of `RuntimeConfig.maxmemory` (`CONFIG SET
+  maxmemory`, server startup) — one Relaxed load, no lock, no per-script
+  Cell bookkeeping.
+- **Lua `redis.call`/`redis.pcall` writes now enforce `--maxmemory`.**
+  EVAL/EVALSHA carry no WRITE command flag, so the dispatch-level OOM check
+  never saw them, and the bridge (`src/scripting/bridge.rs`) never ran the
+  check before executing a write inside a script — a tight `redis.call('SET',
+  ...)` loop could write arbitrarily far past the cap. A new
+  `LuaEvictionCtx`, captured by value into the `redis.call`/`redis.pcall`
+  closures at VM-setup time (`setup_lua_vm`, once per shard — zero new
+  `unsafe`, zero per-call allocation), runs the same gate before a WRITE
+  `redis.call` executes; `redis.call` raises the OOM error as a Lua runtime
+  error (script aborts, matching Redis semantics), `redis.pcall` returns it
+  as an `{err = ...}` table. Read-only scripts are unaffected.
+- **FCALL-internal `redis.call`/`redis.pcall` writes now enforce
+  `--maxmemory` too.** `FUNCTION LOAD` (`src/scripting/functions.rs`)
+  creates its own per-library sandboxed Lua VM, separate from the shard's
+  shared EVAL/EVALSHA VM — that per-library VM registered `redis.call`/
+  `redis.pcall` with `LuaEvictionCtx::disabled()` unconditionally, so a
+  `FUNCTION` whose body wrote in a loop could grow memory past the cap
+  independent of the EVAL/EVALSHA fix above. `FunctionRegistry::new` now
+  takes a `LuaEvictionCtx`, stored on the registry and cloned into every
+  library's VM at `create_library` time; both production construction sites
+  (`handler_monoio/mod.rs`, `handler_sharded/mod.rs`) build a real ctx from
+  the same shard handles `setup_lua_vm` already uses for the shared VM
+  (`ConnectionContext::{shard_databases, runtime_config, shard_id,
+  spill_sender, spill_file_id, disk_offload_dir}`); only test-only callers
+  pass `LuaEvictionCtx::disabled()`.
+- **Cross-shard `MOVE`/`COPY ... DB n` two-database intercept now runs on
+  EVERY `ShardMessage` SPSC arm, not just the plain `Execute` arm.** The
+  intercept (special-cased ahead of the generic single-db write path because
+  both commands need two `&mut Database` borrows at once) previously existed
+  only in `Execute`. Every other arm — including `PipelineBatchSlotted`, the
+  one real client traffic (`handler_monoio`/`handler_sharded`'s pipelined
+  remote-write path) actually uses — fell through to the generic single-db
+  `key_extra::copy`, which parses and silently ignores the `DB` clause: a
+  remote `COPY src dst DB n` performed a same-db copy instead (data
+  corruption, confirmed pre-fix via manual probe: 4/20 remote `COPY`s landed
+  in the wrong db), and a remote `MOVE` returned a loud-but-wrong "MOVE
+  requires handler-level dispatch (cross-db operation)" error instead of
+  moving the key. The intercept is now extracted into a shared helper
+  (`src/shard/spsc_two_db::try_two_db_intercept`) that `Execute`,
+  `MultiExecute`, `PipelineBatch`, `ExecuteSlotted`, `MultiExecuteSlotted`
+  and `PipelineBatchSlotted` all call verbatim; cross-db `COPY` still runs
+  `spsc_eviction_gate` against the destination db before `copy_core` (`COPY`
+  duplicates the value, so it needs the gate; same-shard `MOVE` is net-zero
+  and stays ungated, same rationale as `DEL`).
+- **Found, NOT fixed (out of scope): cross-shard `COPY`'s destination key is
+  only reliably retrievable when it hash-tag-colocates with the source key.**
+  Moon shards purely by key hash, with no db-awareness. A `COPY src dst DB n`
+  executes on whichever shard owns `src` (that's the routing key for the
+  whole command) and writes `dst` into that SAME shard's db space — it does
+  not (and cannot, without a real cross-shard two-phase protocol) relocate
+  the write to whichever shard `dst`'s OWN hash would naturally pick. A later
+  plain `GET dst` routes by `hash(dst)`, so it only reliably finds the value
+  when `src` and `dst` share a hash tag (or `--shards 1`). `MOVE` is exempt
+  (the key name is unchanged, so `hash(key)` is identical before and after).
+  This is a pre-existing, inherent property of the per-key-hash sharding
+  model applied to `COPY` with a renamed destination — the plain `Execute`
+  arm this fix mirrors has the identical property — not a regression from
+  extending the intercept to every arm. Fixing it is a materially larger
+  change (routing the destination write to `dst`'s own natural shard) than
+  "apply the existing intercept to more arms" and is out of scope here.
+- **Found, NOT fixed (out of scope): Moon's eviction gate is per-DATABASE,
+  not per-shard-aggregate-across-databases.** `try_evict_if_needed_budget`
+  (and every SPSC write arm, including the new cross-db `COPY` gate above)
+  measures `db.estimated_memory()` for the ONE database being written to,
+  not the sum across all 16 logical dbs a shard holds — consistent
+  throughout Moon's write path, but a deviation from real Redis's
+  instance-wide `maxmemory`. A cross-db `COPY` into a mostly-empty
+  destination db can undercount total shard pressure. Uniform, pre-existing
+  behavior; fixing it is a cross-cutting change to the eviction model, not a
+  `COPY`/`MOVE`-specific one — out of scope here.
+- New wire-level regression suite `tests/spsc_two_db.rs` (3 cases, both
+  runtimes) exercises the one arm real client traffic uses
+  (`PipelineBatchSlotted`): pipelined cross-shard `COPY ... DB n` (RED
+  before the fix — `MOVE`'s loud wrong error and `COPY`'s silent same-db
+  copy — both confirmed via a git-stash re-run against the pre-fix source),
+  pipelined cross-shard `MOVE`, and a same-db `COPY` control (routes through
+  the coordinator's multi-key scatter path, unaffected by this fix, kept as
+  a negative control). The `COPY` case uses `{i}`-hash-tagged src/dst key
+  pairs so the destination key is independently `GET`-able post-fix — see
+  the "destination hash-tag" caveat above; an untagged version of the same
+  test only succeeds for the ~1-in-4 keys where `hash(dst)` collides with
+  `hash(src)` by chance.
+- New wire-level regression suite `tests/oom_bypass_closure.rs` (8 cases,
+  both runtimes): direct-SET control (A), cross-shard pipeline (B), Lua EVAL
+  loop (C), read-only EVAL under pressure not blocked (D), cross-shard COPY
+  under pressure (E), `CONFIG SET maxmemory` atomic publish/un-publish (F),
+  FCALL-internal write loop (G), FCALL control with no maxmemory (H).
+  B, C, E and G RED before their respective fixes. D locks the write-only
+  scope of the Lua gate. Case F (added with the atomic-snapshot perf
+  follow-up) starts the server with `--maxmemory 0 --disk-offload disable`
+  (both required so the atomic starts genuinely unset — omitting
+  `--maxmemory` trips the auto-guardrail's nonzero default, and
+  disk-offload's spill-sender presence ORs into the same gate) and drives a
+  cross-shard pipeline before and after `CONFIG SET maxmemory`/`CONFIG SET
+  maxmemory 0`; RED before the CONFIG SET call site published the atomic
+  (775/3000 OOM — the local-only share) GREEN after (3000/3000). Case G
+  (FCALL) RED before the FunctionRegistry fix — a 2000-iteration
+  `redis.call('SET', ...)` loop inside an FCALL'd function completed with
+  `'DONE'` instead of an OOM error against a 2MB-capped `noeviction` server;
+  GREEN after. Case H locks the write-only/OOM-only scope of the FCALL gate,
+  mirroring case D for EVAL. Case E was revised for the cross-shard SPSC
+  intercept fix above: it now pre-loads db 1 (the `COPY` destination) with
+  the same ballast as db 0 before the OOM phase, since Moon's per-database
+  eviction gate (see the "found, not fixed" note above) needs the
+  destination db under its OWN pressure to trip — the original version
+  passed only as a side effect of the pre-fix same-db-copy bug (0/300 OOM
+  with the destination-db gate disabled and routing intact, confirmed via a
+  targeted stash of just that gate block; 76/300 fully fixed).
+
 ### Added — int8 symmetric ADC for SQ8 vector search (task #13)
 
 - New per-candidate integer dot-product path for SQ8 asymmetric distance

src/command/config.rs

@@ -102,6 +102,10 @@ pub fn config_set(runtime_config: &mut RuntimeConfig, args: &[Frame]) -> Frame {
                     runtime_config.maxmemory = v;
                     // Keep the inline write path's lock-free pre-gate in sync.
                     crate::storage::eviction::publish_maxmemory_hints(&*runtime_config);
+                    // Keep the SPSC-drain / Lua-bridge "is maxmemory set?" atomic
+                    // in sync (Gap C) — a missed publish here silently bypasses
+                    // the eviction gate on both paths.
+                    crate::storage::eviction::publish_maxmemory(v as u64);
                 }
                 Err(_) => {
                     return Frame::Error(Bytes::from(format!(

src/config.rs

@@ -1024,10 +1024,26 @@ fn detect_memory_limit_bytes() -> Option<usize> {
     }
 }
 
-/// Non-Linux: no portable, dependency-free memory-limit probe. The guardrail
-/// is skipped (operator sets `--maxmemory` explicitly on dev hosts). Production
-/// targets Linux per the platform policy.
-#[cfg(not(target_os = "linux"))]
+/// macOS (first-class target): probe physical RAM via `sysctl -n hw.memsize`.
+/// A spawned command instead of `sysctlbyname` FFI keeps this free of new
+/// 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> {
+    let out = std::process::Command::new("sysctl")
+        .args(["-n", "hw.memsize"])
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    std::str::from_utf8(&out.stdout).ok()?.trim().parse().ok()
+}
+
+/// Other platforms: no portable, dependency-free memory-limit probe. The
+/// 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> {
     None
 }
@@ -1539,6 +1555,19 @@ mod tests {
         assert_eq!(config.to_runtime_config().maxmemory, 0);
     }
 
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn macos_detects_memory_limit() {
+        // macOS is a first-class target; booting UNLIMITED + noeviction by
+        // default (guardrail Skipped) was RSS/CPU/OOM review item 4 — the
+        // hw.memsize probe must feed the same 80% guardrail as Linux.
+        let detected = detect_memory_limit_bytes();
+        assert!(
+            detected.is_some_and(|b| b > 1 << 30),
+            "hw.memsize probe failed: {detected:?}"
+        );
+    }
+
     #[test]
     fn parse_meminfo_extracts_memtotal_bytes() {
         let sample = "MemTotal:       16384256 kB\nMemFree:    1000 kB\n";

src/main.rs

@@ -873,6 +873,10 @@ fn main() -> anyhow::Result<()> {
     // per-shard hint divides by it) — the inline write path's eviction
     // pre-gate reads these instead of taking the runtime-config lock.
     moon::storage::eviction::publish_maxmemory_hints(&runtime_config_shared.read());
+    // Publish the SPSC-drain / Lua-bridge "is maxmemory set?" atomic (Gap C).
+    // Missing this at startup would silently bypass the eviction gate for
+    // any server launched with --maxmemory (until the first CONFIG SET).
+    moon::storage::eviction::publish_maxmemory(runtime_config_shared.read().maxmemory as u64);
     moon::config::log_maxmemory_sharding(runtime_config_shared.read().maxmemory, num_shards);
     let server_config_shared: std::sync::Arc<moon::config::ServerConfig> =
         { std::sync::Arc::new(config.clone()) };

src/scripting/bridge.rs

@@ -7,11 +7,115 @@
 //! 3. The pointer is cleared immediately after script execution
 
 use std::cell::Cell;
+use std::path::PathBuf;
+use std::rc::Rc;
+use std::sync::Arc;
 
 use mlua::prelude::*;
 
+use crate::config::RuntimeConfig;
 use crate::protocol::Frame;
+use crate::shard::shared_databases::ShardDatabases;
 use crate::storage::engine::StorageEngine;
+use crate::storage::eviction::{
+    try_evict_if_needed_async_spill_budget, try_evict_if_needed_budget,
+};
+use crate::storage::tiered::spill_thread::SpillRequest;
+
+/// Shard context needed to enforce `--maxmemory` eviction before a Lua
+/// `redis.call`/`redis.pcall` WRITE actually mutates the database.
+///
+/// # Why closure-capture instead of a thread-local
+///
+/// The bridge already uses a thread-local raw pointer (`CURRENT_DB`) for the
+/// `Database` itself, because that pointer's target changes on every script
+/// invocation. This context is different: it is the same for every script run
+/// on a given shard for the shard's entire lifetime (the shard's
+/// `ShardDatabases`/`RuntimeConfig`/spill handles never change identity).
+/// `redis.call`/`redis.pcall` are Lua closures created exactly once per shard
+/// by [`crate::scripting::setup_lua_vm`], at a point where the caller already
+/// owns cloneable handles to all of this — so it is captured directly into
+/// the `move` closure. This adds zero new `unsafe` code (the existing
+/// `CURRENT_DB` unsafe deref is untouched) and zero per-call allocation. The
+/// common case (`maxmemory` unset, no spill) is decided by a single Relaxed
+/// load of the process-global [`crate::storage::eviction::maxmemory_is_set`]
+/// atomic, so a tight `redis.call('SET', ...)` loop never takes the
+/// `RuntimeConfig` lock at all in that case.
+#[derive(Clone)]
+pub struct LuaEvictionCtx(Option<LuaEvictionInner>);
+
+#[derive(Clone)]
+struct LuaEvictionInner {
+    shard_databases: Arc<ShardDatabases>,
+    runtime_config: Arc<parking_lot::RwLock<RuntimeConfig>>,
+    shard_id: usize,
+    spill_sender: Option<flume::Sender<SpillRequest>>,
+    spill_file_id: Rc<Cell<u64>>,
+    disk_offload_dir: Option<PathBuf>,
+}
+
+impl LuaEvictionCtx {
+    /// No-op gate. Used by unit tests (no real shard context available).
+    /// Production call sites (Lua EVAL/EVALSHA and Lua FUNCTION/FCALL) must
+    /// build a real ctx via [`LuaEvictionCtx::new`] — see
+    /// `src/shard/conn_accept.rs` and `src/scripting/functions.rs`.
+    pub fn disabled() -> Self {
+        LuaEvictionCtx(None)
+    }
+
+    /// Real gate, built from the shard's own handles at VM-setup time.
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        shard_databases: Arc<ShardDatabases>,
+        runtime_config: Arc<parking_lot::RwLock<RuntimeConfig>>,
+        shard_id: usize,
+        spill_sender: Option<flume::Sender<SpillRequest>>,
+        spill_file_id: Rc<Cell<u64>>,
+        disk_offload_dir: Option<PathBuf>,
+    ) -> Self {
+        LuaEvictionCtx(Some(LuaEvictionInner {
+            shard_databases,
+            runtime_config,
+            shard_id,
+            spill_sender,
+            spill_file_id,
+            disk_offload_dir,
+        }))
+    }
+
+    /// Run the same eviction/OOM gate the connection handlers use
+    /// (`run_write_eviction_gate` in `handler_monoio/mod.rs`), against `db`
+    /// (the shard's `Database`, already borrowed by the caller via the
+    /// `CURRENT_DB` thread-local). Returns the standard OOM `Frame::Error`
+    /// on failure; `Ok(())` if within budget, eviction succeeded, or
+    /// `maxmemory` is unset.
+    fn gate(&self, db: &mut crate::storage::Database, db_index: usize) -> Result<(), Frame> {
+        let Some(inner) = self.0.as_ref() else {
+            return Ok(());
+        };
+        // Lock-free fast path: a script issuing thousands of writes checks
+        // the process-global atomic (Gap C), not the RuntimeConfig lock.
+        if inner.spill_sender.is_none() && !crate::storage::eviction::maxmemory_is_set() {
+            return Ok(());
+        }
+        let rt = inner.runtime_config.read();
+        let budget = inner.shard_databases.elastic_budget(inner.shard_id);
+        if let Some(sender) = &inner.spill_sender {
+            let mut fid = inner.spill_file_id.get();
+            let dir = inner
+                .disk_offload_dir
+                .as_deref()
+                .unwrap_or(std::path::Path::new("."));
+            let res = try_evict_if_needed_async_spill_budget(
+                db, &rt, sender, dir, &mut fid, db_index, budget,
+            );
+            inner.spill_file_id.set(fid);
+            res
+        } else {
+            try_evict_if_needed_budget(db, &rt, budget)
+        }
+    }
+}
 
 thread_local! {
     /// Raw pointer to the current shard's Database during script execution.
@@ -59,7 +163,11 @@ pub fn script_had_write() -> bool {
 ///
 /// If `propagate_errors` is true (redis.call), Frame::Error results are raised as Lua errors.
 /// If false (redis.pcall), errors are returned as {err = "..."} tables.
-pub fn make_redis_call_fn(lua: &Lua, propagate_errors: bool) -> mlua::Result<LuaFunction> {
+pub fn make_redis_call_fn(
+    lua: &Lua,
+    propagate_errors: bool,
+    eviction_ctx: LuaEvictionCtx,
+) -> mlua::Result<LuaFunction> {
     lua.create_function(move |lua, args: LuaMultiValue| {
         // Convert all Lua arguments to Frames
         let frames: Vec<Frame> = args
@@ -110,6 +218,14 @@ pub fn make_redis_call_fn(lua: &Lua, propagate_errors: bool) -> mlua::Result<Lua
             if cmd_is_write {
                 // Track writes for SCRIPT KILL safety check
                 SCRIPT_HAD_WRITE.with(|c| c.set(true));
+                // OOM eviction gate (M3): mirrors the connection handlers'
+                // `run_write_eviction_gate` — without this, a write inside a
+                // script could grow memory past `maxmemory` without limit
+                // (EVAL/EVALSHA carry no WRITE command flag, so the
+                // dispatch-level OOM check never sees them at all).
+                if let Err(oom) = eviction_ctx.gate(db, db_idx) {
+                    return Ok(oom);
+                }
             }
 
             let frame = db.execute_command(&cmd_bytes, &frames[1..], &mut db_idx, db_count);