perf(replication): per-write fanout gate — parking_lot migration + FANOUT_HINT probe fix + dead PSYNC path deletion (tasks #70, #72)

CHANGELOG.md

@@ -38,6 +38,115 @@ overwhelmingly common case — is unaffected and stays recyclable.
 Byte-slice peek only, no allocation and no LZ4 decompression (`Command`
 payloads are never LZ4-compressed by `write_wal_v3_record`) — off the hot
 path (recycle passes only) but kept allocation-free per repo convention.
+### Fixed — Replication fanout gate perf debt: `parking_lot` migration, FANOUT_HINT mis-activation, single-guard write path (task #70); deleted dead master-side PSYNC path (task #72)
+
+Two pre-soak perf/hygiene defects blocking the v0.7.0 tag, fixed together
+since both touch `ReplicationState` locking on the per-write hot path.
+
+**Task #70 (4 sub-fixes):**
+
+1. `ReplicationState`'s outer lock (`ConnectionContext::repl_state` and every
+   `Arc<RwLock<ReplicationState>>` holder — 27 call sites across
+   `admin/metrics_setup.rs`, `cluster/gossip.rs`, `command/server_admin.rs`,
+   `main.rs`, `persistence/aof/pool.rs`, `replication/{reason_del,replica,
+   master,state}.rs`, `scripting/bridge.rs`, `server/conn/*`, `shard/*`) was
+   `std::sync::RwLock` while the inner `per_shard_backlogs` was already
+   `parking_lot::Mutex` — an inconsistent locking policy and a source of
+   `.read().unwrap()` / `if let Ok(g) = rs.read()` poisoning dances on the
+   write path. Migrated the outer lock to `parking_lot::RwLock` throughout;
+   removed all poisoning-related `Result` unwrapping (parking_lot's
+   `read()`/`write()` return the guard directly, `try_read()`/`try_write()`
+   return `Option`, not `Result`).
+2. `ensure_backlogs_allocated()` (called from `try_handle_replconf` on ANY
+   bare `REPLCONF`, including monitoring probes and failed handshakes) used
+   to call `mark_fanout_active()` unconditionally. Since `FANOUT_HINT` is a
+   sticky, never-cleared process-global flag, a single stray `REPLCONF`
+   permanently taxed every subsequent write with the fanout-active gate
+   check — even on a server that never completes a PSYNC. Split allocation
+   from hint-activation: `ensure_backlogs_allocated()` still allocates
+   backlogs (load-bearing for the handshake) but no longer touches
+   `FANOUT_HINT`; `mark_fanout_active()` now fires only from
+   `try_handle_psync`, on an actual PSYNC/replica registration.
+3. `record_local_write` / `record_local_write_db` took up to 3-4 separate
+   `repl_state.read()` acquisitions per write (SELECT-needed check, backlog
+   append, offset advance, ...). Collapsed the internal chain to ONE guard
+   acquired at the top of `record_local_write_db` and threaded through to
+   `record_local_write` (now `fn record_local_write(g: &ReplicationState,
+   shard_id, bytes)`, no longer self-locking). `replication_fanout_active`
+   stays a separate, short-lived gate lock — deliberately NOT folded into the
+   same guard, because some external call sites (`handler_monoio/mod.rs`'s
+   MOVE/COPY handling) reach an `.await` between the gate check and the next
+   repl-state touch, and Rust drops lock guards at end of lexical scope, not
+   at last use — holding one guard across that span would violate the
+   never-lock-across-`.await` rule. Net effect: at most 2 lock acquisitions
+   per replicated write (1 gate + 1 write guard), down from up to 4-5.
+4. Added `#[inline]` to `record_local_write`, `record_local_write_db`, and
+   `replication_fanout_active`, matching the sibling `try_handle_*`
+   convention in `dispatch.rs`.
+5. **Correctness follow-up (review-caught regression on #2):** splitting
+   backlog allocation from hint activation opened a REPLCONF→PSYNC window
+   where a bare `REPLCONF` could allocate + seed a shard's
+   `ReplicationBacklog` at its offset at that instant, then — with
+   `FANOUT_HINT` still false — every subsequent local write advanced the
+   shard's real offset counter via the bare-`issue_lsn` branch with NO
+   matching backlog append. The backlog's `end_offset` silently skewed stale
+   relative to the real counter for as long as the window stayed open (an
+   AOF-enabled master under continuous write load with periodic replica
+   kill-9/reconnect — exactly the v0.7.0 24h soak's shape). Any
+   snapshot/cut/push-offset captured from that backlog after activation
+   would then be range-inconsistent with it, corrupting partial-resync
+   catch-up reads (wrong bytes or a silently-skipped catch-up) and
+   acked-write accounting. Fixed by adding
+   `ReplicationBacklog::realign_to(offset)` (resets the buffer to empty,
+   reseeded at `offset` — a skewed buffer's contents are already useless, so
+   dropping them is correct; an offset request that falls below the new
+   `start_offset` fails safe into a full resync) and
+   `ReplicationState::realign_backlog(shard_id)`, called at every activation
+   site — `try_handle_psync` (dispatch.rs, right after `mark_fanout_active`)
+   and the `RegisterReplica`/`PrepareReplicaSync` arms
+   (`shard/spsc_handler.rs`) — BEFORE any snapshot/cut/push offset is
+   captured. All three sites run on the shard thread that owns that shard's
+   own offset advances, so the offset read + realign is race-free with the
+   shard's own append/advance sequence.
+
+**Task #72:** `handle_psync_on_master` (both `runtime-tokio` and
+`runtime-monoio` variants) and `register_replica_with_shards` in
+`src/replication/master.rs` were unreachable — every PSYNC handler actually
+wired from `shard/conn_accept.rs` goes through
+`handle_psync_inline_single_shard` / `handle_psync_inline_multi_shard`
+instead. The dead pair also carried a latent broken WAIT implementation: it
+initialized `ack_offsets` but never spawned the `ack_read_loop` that drains
+replica ACKs into them, so `wait_for_replicas` against a registration made
+through that path would have blocked forever. Deleted both variants plus
+`evaluate_psync_shared` (only reachable from the deleted code); kept
+`backlog_bytes_from`, still used by the live `send_backlog_range`.
+
+Two new unit tests (`replication::state::tests`) cover the FANOUT_HINT fix
+with delta-based assertions (robust against shared-binary test-order
+contamination): `test_ensure_backlogs_allocated_does_not_activate_fanout_hint`
+(RED on the pre-fix code — verified by temporarily reintroducing the bare
+`mark_fanout_active()` call and observing the assertion fail) and
+`test_mark_fanout_active_sets_hint`. Four more cover the backlog-realign
+correctness fix (#70.5): `test_realign_to_resets_skewed_backlog` and
+`test_realign_to_noop_when_already_aligned` (`replication::backlog::tests`,
+the low-level buffer reset behavior) plus
+`test_realign_backlog_fixes_skew_after_hint_false_window` and
+`test_realign_backlog_then_append_reads_back_exact_record`
+(`replication::state::tests`, reproducing the exact skew scenario — allocate,
+advance the offset via `issue_lsn` with no append, then activate — and
+proving the post-fix backlog is byte-position-consistent with the real
+offset again). RED verified by temporarily neutering `realign_backlog`'s
+body and observing both `state::tests` cases fail (`end_offset` stuck at 0
+instead of matching the real offset 300; the post-append record unreadable
+at its own pre-append offset), then reverting.
+
+No wire-protocol or observable behavior change beyond the hint gating: the
+REPLCONF → PSYNC handshake sequence is unchanged and covered end-to-end by
+the `replication_hardening` / `replication_multishard` / `replication_planes`
+integration suites (all green, monoio — master-side PSYNC is monoio-only by
+pre-existing design, `try_handle_psync_unsupported` under tokio, untouched
+by this change) plus the full `replication::` unit-test module under both
+`runtime-monoio` (default) and `runtime-tokio,jemalloc`.
 
 ### Fixed — `crash_recovery_disk_offload_no_aof` harness assumed eviction-throughput durability the write path no longer provides (task #44)
 

src/admin/metrics_setup.rs

@@ -1238,54 +1238,54 @@ pub fn get_cpu_usage() -> (f64, f64) {
 // ── Global replication state (for INFO) ────────────────────────────────
 
 static GLOBAL_REPL_STATE: once_cell::sync::OnceCell<
-    std::sync::Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>,
+    std::sync::Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>,
 > = once_cell::sync::OnceCell::new();
 
 /// Register the global replication state for INFO queries.
 pub fn set_global_repl_state(
-    state: std::sync::Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>,
+    state: std::sync::Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>,
 ) {
     let _ = GLOBAL_REPL_STATE.set(state);
 }
 
 /// Get the raw global replication state Arc (for MEMORY DOCTOR backlog query).
 /// Returns None before replication is initialized.
 pub fn get_global_repl_state_arc()
--> Option<&'static std::sync::Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>> {
+-> Option<&'static std::sync::Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>>
+{
     GLOBAL_REPL_STATE.get()
 }
 
 /// Get replication info for INFO command: (role, connected_slaves, master_repl_offset, repl_id).
 /// Also updates the Prometheus replication lag gauge as a side-effect.
 pub fn get_replication_info() -> (&'static str, usize, u64, String) {
     if let Some(state) = GLOBAL_REPL_STATE.get() {
-        if let Ok(guard) = state.read() {
-            let role = match &guard.role {
-                crate::replication::state::ReplicationRole::Master => "master",
-                crate::replication::state::ReplicationRole::Replica { .. } => "slave",
-            };
-            let slaves = guard.replicas.len();
-            let offset = guard.master_repl_offset.load(Ordering::Relaxed);
-            let repl_id = guard.repl_id.clone();
-            // Update Prometheus lag gauge: max lag across all replicas.
-            if !guard.replicas.is_empty() {
-                let max_lag_bytes = guard
-                    .replicas
-                    .iter()
-                    .map(|r| {
-                        let ack: u64 = r
-                            .ack_offsets
-                            .iter()
-                            .map(|a| a.load(Ordering::Relaxed))
-                            .sum();
-                        offset.saturating_sub(ack)
-                    })
-                    .max()
-                    .unwrap_or(0);
-                record_replication_lag(max_lag_bytes, 0);
-            }
-            return (role, slaves, offset, repl_id);
+        let guard = state.read();
+        let role = match &guard.role {
+            crate::replication::state::ReplicationRole::Master => "master",
+            crate::replication::state::ReplicationRole::Replica { .. } => "slave",
+        };
+        let slaves = guard.replicas.len();
+        let offset = guard.master_repl_offset.load(Ordering::Relaxed);
+        let repl_id = guard.repl_id.clone();
+        // Update Prometheus lag gauge: max lag across all replicas.
+        if !guard.replicas.is_empty() {
+            let max_lag_bytes = guard
+                .replicas
+                .iter()
+                .map(|r| {
+                    let ack: u64 = r
+                        .ack_offsets
+                        .iter()
+                        .map(|a| a.load(Ordering::Relaxed))
+                        .sum();
+                    offset.saturating_sub(ack)
+                })
+                .max()
+                .unwrap_or(0);
+            record_replication_lag(max_lag_bytes, 0);
         }
+        return (role, slaves, offset, repl_id);
     }
     ("master", 0, 0, "0".repeat(40))
 }
@@ -1456,9 +1456,7 @@ fn update_moon_memory_bytes() {
 
     // Replication backlog via global state.
     if let Some(state) = get_global_repl_state_arc() {
-        if let Ok(guard) = state.read() {
-            backlog = guard.backlog_resident_bytes();
-        }
+        backlog = state.read().backlog_resident_bytes();
     }
 
     // task #58 (LOW-1): read the allocator-overhead figure sampled by shard

src/cluster/gossip.rs

@@ -360,7 +360,7 @@ pub async fn run_gossip_ticker(
     node_timeout_ms: u64,
     shutdown: CancellationToken,
     vote_tx: SharedVoteTx,
-    repl_state: std::sync::Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>,
+    repl_state: std::sync::Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>,
 ) {
     let mut tick = tokio::time::interval(Duration::from_millis(100));
     let mut election_spawned = false;
@@ -402,7 +402,7 @@ pub async fn run_gossip_ticker(
                                 }
                                 let cs_election = cluster_state.clone();
                                 let sa = self_addr;
-                                let offset = repl_state.read().unwrap().total_offset();
+                                let offset = repl_state.read().total_offset();
                                 let vtx = vote_tx.clone();
                                 tokio::spawn(async move {
                                     crate::cluster::failover::run_election_task(
@@ -494,7 +494,7 @@ pub async fn run_gossip_ticker(
     node_timeout_ms: u64,
     shutdown: CancellationToken,
     vote_tx: SharedVoteTx,
-    repl_state: std::sync::Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>,
+    repl_state: std::sync::Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>,
 ) {
     let mut election_spawned = false;
     // Bound outstanding PING probes (see the tokio variant): one probe at a
@@ -530,7 +530,7 @@ pub async fn run_gossip_ticker(
                                 }
                                 let cs_election = cluster_state.clone();
                                 let sa = self_addr;
-                                let offset = repl_state.read().unwrap().total_offset();
+                                let offset = repl_state.read().total_offset();
                                 let vtx = vote_tx.clone();
                                 monoio::spawn(async move {
                                     crate::cluster::failover::run_election_task(

src/command/server_admin.rs

@@ -635,9 +635,7 @@ fn days_to_ymd(days: u64) -> (u64, u64, u64) {
 /// Read replication backlog resident bytes via the global state.
 fn replication_backlog_bytes() -> usize {
     if let Some(state) = crate::admin::metrics_setup::get_global_repl_state_arc() {
-        if let Ok(guard) = state.read() {
-            return guard.backlog_resident_bytes();
-        }
+        return state.read().backlog_resident_bytes();
     }
     0
 }

src/main.rs

@@ -894,7 +894,7 @@ fn main() -> anyhow::Result<()> {
     let repl_state = {
         let mut rs = moon::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2);
         rs.set_backlog_capacity(config.repl_backlog_size);
-        std::sync::Arc::new(std::sync::RwLock::new(rs))
+        std::sync::Arc::new(parking_lot::RwLock::new(rs))
     };
 
     // Register repl_state globally for INFO command queries.
@@ -1382,10 +1382,8 @@ fn main() -> anyhow::Result<()> {
                 // RFC § 2 Rule 3 — seed master_repl_offset before accepting
                 // client traffic so the next write doesn't reissue an LSN
                 // already on disk.
-                if global_max_lsn > 0
-                    && let Ok(state) = repl_state.read()
-                {
-                    state.seed_master_offset(global_max_lsn);
+                if global_max_lsn > 0 {
+                    repl_state.read().seed_master_offset(global_max_lsn);
                 }
 
                 // Retire any stray legacy top-level appendonly.aof so the

src/persistence/aof/pool.rs

@@ -734,10 +734,9 @@ impl AofWriterPool {
     /// `ReplicationState::issue_lsn` so handler call sites collapse to a
     /// single line.
     ///
-    /// Returns 0 when:
-    /// - `repl_state` is None (test fixtures or shutdown paths)
-    /// - the `RwLock` is poisoned (shouldn't happen in production —
-    ///   ReplicationState is only `write()`-locked under known-safe paths)
+    /// Returns 0 when `repl_state` is None (test fixtures or shutdown paths).
+    /// `parking_lot::RwLock` does not poison, so there is no lock-error case
+    /// to fall back on here (task #70 — was a std::sync poisoning dance).
     ///
     /// 0 is a sentinel meaning "no replication ordering for this write".
     /// TopLevel writers ignore the LSN entirely so 0 is harmless there;
@@ -746,13 +745,13 @@ impl AofWriterPool {
     /// for the cross-shard `OrderedAcrossShards` merge in RFC step 5.
     #[inline]
     pub fn issue_append_lsn(
-        repl_state: &Option<Arc<std::sync::RwLock<crate::replication::state::ReplicationState>>>,
+        repl_state: &Option<Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>>,
         shard_id: usize,
         delta: usize,
     ) -> u64 {
         repl_state
             .as_ref()
-            .and_then(|rs| rs.read().ok().map(|g| g.issue_lsn(shard_id, delta as u64)))
+            .map(|rs| rs.read().issue_lsn(shard_id, delta as u64))
             .unwrap_or(0)
     }