diff --git a/CHANGELOG.md b/CHANGELOG.md index a0417f79..3df02db8 100644 --- a/CHANGELOG.md +++ b/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>` 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) diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index 0efeb817..6f5fb72a 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -1238,12 +1238,12 @@ 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::Arc>, > = once_cell::sync::OnceCell::new(); /// Register the global replication state for INFO queries. pub fn set_global_repl_state( - state: std::sync::Arc>, + state: std::sync::Arc>, ) { let _ = GLOBAL_REPL_STATE.set(state); } @@ -1251,7 +1251,8 @@ pub fn set_global_repl_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>> { +-> Option<&'static std::sync::Arc>> +{ GLOBAL_REPL_STATE.get() } @@ -1259,33 +1260,32 @@ pub fn get_global_repl_state_arc() /// 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 diff --git a/src/cluster/gossip.rs b/src/cluster/gossip.rs index cad6ee54..41da2b38 100644 --- a/src/cluster/gossip.rs +++ b/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>, + repl_state: std::sync::Arc>, ) { 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>, + repl_state: std::sync::Arc>, ) { 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( diff --git a/src/command/server_admin.rs b/src/command/server_admin.rs index fa43f392..5745da85 100644 --- a/src/command/server_admin.rs +++ b/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 } diff --git a/src/main.rs b/src/main.rs index 36e683b1..abe17162 100644 --- a/src/main.rs +++ b/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 diff --git a/src/persistence/aof/pool.rs b/src/persistence/aof/pool.rs index b0d5fde0..7f284b76 100644 --- a/src/persistence/aof/pool.rs +++ b/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>>, + repl_state: &Option>>, 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) } diff --git a/src/replication/backlog.rs b/src/replication/backlog.rs index 250a8f48..ccfd138c 100644 --- a/src/replication/backlog.rs +++ b/src/replication/backlog.rs @@ -83,6 +83,35 @@ impl ReplicationBacklog { offset >= self.start_offset && offset <= self.end_offset } + /// Reset the backlog to empty, seeded at `offset` — the same state as + /// `new_at(capacity, offset)`. No-op if already aligned + /// (`end_offset == offset`). + /// + /// Task #70 correctness fix: an allocated-but-idle backlog (seeded by a + /// bare `REPLCONF` via `ensure_backlogs_allocated`) can sit at a stale + /// offset X while the shard's real offset counter keeps advancing to Y + /// via `issue_lsn` — the FANOUT_HINT gate being false means those writes + /// take the bare-LSN branch with NO backlog append. Once activation + /// happens (an actual PSYNC, or a shard's own RegisterReplica / + /// PrepareReplicaSync arm), the backlog must be realigned to the shard's + /// CURRENT offset before any snapshot/cut offset is captured from it — + /// otherwise `bytes_from`/`contains_offset` answer against the wrong + /// byte positions (skew Y−X), corrupting partial-resync and live-fanout + /// cut math. A skewed buffer's contents are already useless (they cover + /// bytes that were never appended for the real offset range), so + /// dropping them is correct: any partial-resync request for an offset + /// below the new `start_offset` falls below the window and + /// `bytes_from` returns `None` — the caller's existing fail-safe is a + /// full resync, never wrong bytes. + pub fn realign_to(&mut self, offset: u64) { + if self.end_offset == offset { + return; + } + self.buf.clear(); + self.start_offset = offset; + self.end_offset = offset; + } + pub fn start_offset(&self) -> u64 { self.start_offset } @@ -199,4 +228,39 @@ mod tests { assert!(s3 >= s2, "start_offset must be monotonic"); assert!(e3 > e2, "end_offset must be strictly increasing on append"); } + + /// Task #70 correctness fix: `realign_to` must reset a skewed backlog + /// to a clean, empty state seeded at the given offset — matching + /// `new_at(capacity, offset)` exactly. + #[test] + fn test_realign_to_resets_skewed_backlog() { + let mut bl = ReplicationBacklog::new_at(1024, 100); + bl.append(b"stale bytes nobody wants"); + assert_ne!(bl.end_offset(), 500); + + bl.realign_to(500); + assert_eq!(bl.start_offset(), 500); + assert_eq!(bl.end_offset(), 500); + // The stale bytes are gone: bytes_from(the old start) must not + // return them. + assert_eq!(bl.bytes_from(500), Some(vec![])); + + // A fresh append after realign lands at the NEW offset, not the old one. + bl.append(b"fresh"); + assert_eq!(bl.start_offset(), 500); + assert_eq!(bl.end_offset(), 505); + assert_eq!(bl.bytes_from(500), Some(b"fresh".to_vec())); + } + + /// Realigning to the CURRENT end_offset (no skew) must be a true no-op: + /// existing buffered bytes are preserved, not dropped. + #[test] + fn test_realign_to_noop_when_already_aligned() { + let mut bl = ReplicationBacklog::new(1024); + bl.append(b"hello"); + bl.realign_to(5); // end_offset is already 5 + assert_eq!(bl.start_offset(), 0); + assert_eq!(bl.end_offset(), 5); + assert_eq!(bl.bytes_from(0), Some(b"hello".to_vec())); + } } diff --git a/src/replication/master.rs b/src/replication/master.rs index ca08aaf0..a29a9ca1 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -1,612 +1,47 @@ //! Master-side PSYNC2 handler and WAIT command support. //! -//! Provides `handle_psync_on_master` for incoming PSYNC connections -//! and `wait_for_replicas` for the WAIT command. -#![allow(unused_imports)] +//! Provides the inline single-/multi-shard PSYNC handlers for incoming PSYNC +//! connections (monoio-only — tokio rejects master-side PSYNC upstream via +//! `try_handle_psync_unsupported`) and `wait_for_replicas` for the WAIT +//! command. +//! +//! The original cross-shard-coordinated `handle_psync_on_master` + +//! `register_replica_with_shards` pair (both tokio and monoio variants) was +//! dead code — never called from any handler — and was deleted (task #72). +//! Its WAIT-ack wiring was also latent-broken: it initialized `ack_offsets` +//! but never spawned `ack_read_loop` to drain replica ACKs into them, so +//! `wait_for_replicas` against those registrations would never observe a +//! non-zero ack. The live path (`handle_psync_inline_single_shard` / +//! `handle_psync_inline_multi_shard`, called from +//! `shard/conn_accept.rs`) does this correctly. + +use std::sync::Arc; -use std::sync::{Arc, RwLock}; +use parking_lot::RwLock; #[cfg(feature = "runtime-monoio")] use std::cell::RefCell; #[cfg(feature = "runtime-monoio")] use std::rc::Rc; -#[cfg(feature = "runtime-tokio")] -use tokio::io::AsyncWriteExt; -#[cfg(feature = "runtime-tokio")] -use tokio::net::tcp::OwnedWriteHalf; +#[cfg(feature = "runtime-monoio")] use tracing::info; +#[cfg(feature = "runtime-monoio")] use crate::replication::backlog::SharedBacklog; +#[cfg(feature = "runtime-monoio")] use crate::replication::handshake::PsyncDecision; -use crate::replication::state::{ReplicaInfo, ReplicationState}; - -/// Evaluate PSYNC against shared backlogs by briefly taking each shard's mutex -/// to call `evaluate_psync` against the backlog snapshot. -fn evaluate_psync_shared( - client_repl_id: &str, - client_offset: i64, - server_repl_id: &str, - server_repl_id2: &str, - shared: &[SharedBacklog], -) -> PsyncDecision { - if client_offset < 0 { - return PsyncDecision::FullResync; - } - let id_matches = client_repl_id == server_repl_id || client_repl_id == server_repl_id2; - if !id_matches { - return PsyncDecision::FullResync; - } - let offset = client_offset as u64; - let all_cover = shared.iter().all(|s| { - let g = s.lock(); - g.as_ref().is_some_and(|b| b.contains_offset(offset)) - }); - if all_cover { - PsyncDecision::PartialResync { - from_offset: offset, - } - } else { - PsyncDecision::FullResync - } -} +#[cfg(feature = "runtime-monoio")] +use crate::replication::state::ReplicaInfo; +use crate::replication::state::ReplicationState; /// Read backlog bytes from one shard, returning None if the offset is evicted /// or the backlog is unallocated. +#[cfg(feature = "runtime-monoio")] fn backlog_bytes_from(shared: &SharedBacklog, from_offset: u64) -> Option> { let g = shared.lock(); g.as_ref().and_then(|b| b.bytes_from(from_offset)) } -/// Master-side PSYNC handler: evaluate the request, respond, and wire up replication. -/// -/// Called from handle_connection_sharded when PSYNC arrives on a connection. -/// Returns Ok(()) after handing the connection off to per-shard replica sender tasks. -/// -/// Full resync flow: -/// 1. Record snapshot_start_offset from current master_repl_offset -/// 2. Send SnapshotBegin to ALL shards simultaneously -/// 3. Await all N snapshot completions -/// 4. Send per-shard RDB files as $\r\n\r\n bulk strings -/// 5. Send backlog bytes from snapshot_start_offset to current offset -/// 6. Register replica (RegisterReplica) with all shards for live streaming -/// -/// Partial resync flow: -/// 1. Send +CONTINUE -/// 2. Send backlog bytes from client_offset to current offset for each shard -/// 3. Register replica with all shards for live streaming -#[cfg(feature = "runtime-tokio")] -#[tracing::instrument(skip_all, level = "debug", fields(repl_id = %client_repl_id, offset = client_offset))] -pub async fn handle_psync_on_master( - client_repl_id: &str, - client_offset: i64, - mut write_half: OwnedWriteHalf, - repl_state: Arc>, - per_shard_backlogs: &[SharedBacklog], - shard_producers: &mut Vec>, - persistence_dir: &str, - replica_addr: std::net::SocketAddr, -) -> anyhow::Result<()> { - let (repl_id, repl_id2, current_offset) = { - let rs = repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; - (rs.repl_id.clone(), rs.repl_id2.clone(), rs.total_offset()) - }; - - let decision = evaluate_psync_shared( - client_repl_id, - client_offset, - &repl_id, - &repl_id2, - per_shard_backlogs, - ); - - match decision { - PsyncDecision::FullResync => { - // Respond: +FULLRESYNC - let response = format!("+FULLRESYNC {} {}\r\n", repl_id, current_offset); - write_half.write_all(response.as_bytes()).await?; - - let snapshot_start_offset = current_offset; - - // Trigger per-shard snapshots in parallel - let snap_dir = std::path::PathBuf::from(persistence_dir); - let epoch = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - - let num_shards = shard_producers.len(); - let mut snap_rxs: Vec>> = - Vec::new(); - - for (shard_id, prod) in shard_producers.iter_mut().enumerate() { - use ringbuf::traits::Producer; - let (tx, rx) = crate::runtime::channel::oneshot(); - let msg = crate::shard::dispatch::ShardMessage::SnapshotBegin { - epoch, - snapshot_dir: snap_dir.clone(), - reply_tx: tx, - }; - if prod.try_push(msg).is_err() { - anyhow::bail!("Failed to send SnapshotBegin to shard {}", shard_id); - } - snap_rxs.push(rx); - } - - // Await all shard snapshots - for (shard_id, rx) in snap_rxs.into_iter().enumerate() { - match rx.await { - Ok(Ok(())) => info!("Master: shard {} snapshot complete", shard_id), - Ok(Err(e)) => anyhow::bail!("Shard {} snapshot failed: {}", shard_id, e), - Err(_) => anyhow::bail!("Shard {} snapshot channel dropped", shard_id), - } - } - - // Transfer per-shard RDB files using async I/O to avoid blocking the event loop. - // - // TODO: For standard Redis replicas, convert RRDSHARD data to Redis RDB format - // using crate::persistence::redis_rdb::write_rdb() before sending. Currently we - // send RRDSHARD format which our own replicas understand natively. The redis_rdb - // module (from Plan 43-01) provides the conversion primitives when needed. - for shard_id in 0..num_shards { - let snap_path = snap_dir.join(format!("shard-{}.rrdshard", shard_id)); - let data = tokio::fs::read(&snap_path).await.map_err(|e| { - anyhow::anyhow!( - "Failed to read shard {} snapshot at {:?}: {}", - shard_id, - snap_path, - e - ) - })?; - let header = format!("${}\r\n", data.len()); - write_half.write_all(header.as_bytes()).await?; - write_half.write_all(&data).await?; - write_half.write_all(b"\r\n").await?; - info!("Master: sent shard {} RDB ({} bytes)", shard_id, data.len()); - } - - // Stream backlog bytes accumulated since snapshot_start_offset - for (shard_id, backlog) in per_shard_backlogs.iter().enumerate() { - if let Some(bytes) = backlog_bytes_from(backlog, snapshot_start_offset) { - if !bytes.is_empty() { - write_half.write_all(&bytes).await?; - info!( - "Master: sent shard {} backlog ({} bytes)", - shard_id, - bytes.len() - ); - } - } - } - - // Register this replica with all shards for live WAL streaming - register_replica_with_shards( - replica_addr, - write_half, - repl_state, - shard_producers, - num_shards, - ) - .await?; - } - - PsyncDecision::PartialResync { from_offset } => { - // Respond: +CONTINUE - let response = format!("+CONTINUE {}\r\n", repl_id); - write_half.write_all(response.as_bytes()).await?; - - let num_shards = shard_producers.len(); - - // Stream backlog bytes from from_offset to current for each shard - for (shard_id, backlog) in per_shard_backlogs.iter().enumerate() { - if let Some(bytes) = backlog_bytes_from(backlog, from_offset) { - if !bytes.is_empty() { - write_half.write_all(&bytes).await?; - info!( - "Master: partial resync shard {} ({} bytes)", - shard_id, - bytes.len() - ); - } - } - } - - // Register for live streaming - register_replica_with_shards( - replica_addr, - write_half, - repl_state, - shard_producers, - num_shards, - ) - .await?; - } - } - - Ok(()) -} - -/// Master-side PSYNC handler for monoio runtime. -/// -/// Same logic as the tokio variant but uses monoio ownership I/O for all TCP writes. -/// Takes a mutable reference to `monoio::net::TcpStream` instead of `OwnedWriteHalf`. -#[cfg(feature = "runtime-monoio")] -#[tracing::instrument(skip_all, level = "debug", fields(repl_id = %client_repl_id, offset = client_offset))] -pub async fn handle_psync_on_master( - client_repl_id: &str, - client_offset: i64, - mut stream: monoio::net::TcpStream, - repl_state: Arc>, - per_shard_backlogs: &[SharedBacklog], - shard_producers: &mut Vec>, - persistence_dir: &str, - replica_addr: std::net::SocketAddr, -) -> anyhow::Result<()> { - use monoio::io::AsyncWriteRentExt; - - let (repl_id, repl_id2, current_offset) = { - let rs = repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; - (rs.repl_id.clone(), rs.repl_id2.clone(), rs.total_offset()) - }; - - let decision = evaluate_psync_shared( - client_repl_id, - client_offset, - &repl_id, - &repl_id2, - per_shard_backlogs, - ); - - match decision { - PsyncDecision::FullResync => { - // Respond: +FULLRESYNC - let response = format!("+FULLRESYNC {} {}\r\n", repl_id, current_offset); - let data = response.into_bytes(); - let (wr, _) = stream.write_all(data).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - - let snapshot_start_offset = current_offset; - - // Trigger per-shard snapshots in parallel - let snap_dir = std::path::PathBuf::from(persistence_dir); - let epoch = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - - let num_shards = shard_producers.len(); - let mut snap_rxs: Vec>> = - Vec::new(); - - for (shard_id, prod) in shard_producers.iter_mut().enumerate() { - use ringbuf::traits::Producer; - let (tx, rx) = crate::runtime::channel::oneshot(); - let msg = crate::shard::dispatch::ShardMessage::SnapshotBegin { - epoch, - snapshot_dir: snap_dir.clone(), - reply_tx: tx, - }; - if prod.try_push(msg).is_err() { - anyhow::bail!("Failed to send SnapshotBegin to shard {}", shard_id); - } - snap_rxs.push(rx); - } - - // Await all shard snapshots - for (shard_id, rx) in snap_rxs.into_iter().enumerate() { - match rx.await { - Ok(Ok(())) => info!("Master: shard {} snapshot complete", shard_id), - Ok(Err(e)) => anyhow::bail!("Shard {} snapshot failed: {}", shard_id, e), - Err(_) => anyhow::bail!("Shard {} snapshot channel dropped", shard_id), - } - } - - // Transfer per-shard RDB files. - // Monoio: synchronous file read. Thread-per-core model means this - // blocks only this core's event loop. For large files, consider - // monoio::fs::File with read_at() in the future. - // - // TODO: For standard Redis replicas, convert RRDSHARD data to Redis RDB format - // using crate::persistence::redis_rdb::write_rdb() before sending. Currently we - // send RRDSHARD format which our own replicas understand natively. - for shard_id in 0..num_shards { - let snap_path = snap_dir.join(format!("shard-{}.rrdshard", shard_id)); - let file_data = std::fs::read(&snap_path).map_err(|e| { - anyhow::anyhow!( - "Failed to read shard {} snapshot at {:?}: {}", - shard_id, - snap_path, - e - ) - })?; - let header = format!("${}\r\n", file_data.len()); - let (wr, _) = stream.write_all(header.into_bytes()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - let (wr, _) = stream.write_all(file_data).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - let (wr, _) = stream.write_all(b"\r\n".to_vec()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - info!( - "Master: sent shard {} RDB ({} bytes)", - shard_id, - std::fs::metadata(&snap_path).map(|m| m.len()).unwrap_or(0) - ); - } - - // Stream backlog bytes accumulated since snapshot_start_offset - for (shard_id, backlog) in per_shard_backlogs.iter().enumerate() { - if let Some(bytes) = backlog_bytes_from(backlog, snapshot_start_offset) { - if !bytes.is_empty() { - let (wr, _) = stream.write_all(bytes.to_vec()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - info!( - "Master: sent shard {} backlog ({} bytes)", - shard_id, - bytes.len() - ); - } - } - } - - // Register this replica with all shards for live WAL streaming - register_replica_with_shards( - replica_addr, - stream, - repl_state, - shard_producers, - num_shards, - ) - .await?; - } - - PsyncDecision::PartialResync { from_offset } => { - // Respond: +CONTINUE - let response = format!("+CONTINUE {}\r\n", repl_id); - let (wr, _) = stream.write_all(response.into_bytes()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - - let num_shards = shard_producers.len(); - - // Stream backlog bytes from from_offset to current for each shard - for (shard_id, backlog) in per_shard_backlogs.iter().enumerate() { - if let Some(bytes) = backlog_bytes_from(backlog, from_offset) { - if !bytes.is_empty() { - let (wr, _) = stream.write_all(bytes.to_vec()).await; - wr.map_err(|e| anyhow::anyhow!(e))?; - info!( - "Master: partial resync shard {} ({} bytes)", - shard_id, - bytes.len() - ); - } - } - } - - // Register for live streaming - register_replica_with_shards( - replica_addr, - stream, - repl_state, - shard_producers, - num_shards, - ) - .await?; - } - } - - Ok(()) -} - -/// Assign a unique replica ID and register the replica's write half with all shards. -/// -/// For each shard, creates a bounded mpsc channel, spawns a replica_sender_task -/// that drains the channel to the socket, and sends RegisterReplica to each shard. -#[cfg(feature = "runtime-tokio")] -async fn register_replica_with_shards( - addr: std::net::SocketAddr, - write_half: OwnedWriteHalf, - repl_state: Arc>, - shard_producers: &mut Vec>, - num_shards: usize, -) -> anyhow::Result<()> { - use ringbuf::traits::Producer; - use std::sync::atomic::Ordering; - - static NEXT_REPLICA_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); - let replica_id = NEXT_REPLICA_ID.fetch_add(1, Ordering::Relaxed); - - // Share the write_half across per-shard sender tasks - let write_half = Arc::new(tokio::sync::Mutex::new(write_half)); - - // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. - let backlog_capacity = repl_state - .read() - .map(|g| g.backlog_capacity) - .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); - - let channel_capacity = 1024; - let mut shard_txs = Vec::with_capacity(num_shards); - let mut ack_offsets = Vec::with_capacity(num_shards); - - for shard_id in 0..num_shards { - let (tx, rx) = crate::runtime::channel::mpsc_bounded::(channel_capacity); - shard_txs.push(tx.clone()); - ack_offsets.push(std::sync::atomic::AtomicU64::new(0)); - - // Send RegisterReplica to the shard's SPSC - if let Some(prod) = shard_producers.get_mut(shard_id) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica(Box::new( - crate::shard::dispatch::RegisterReplicaPayload { - replica_id, - tx, - // Legacy multi-shard drain loops do not poll the kick flag - // (superseded by the R2 redesign); overflow still stops - // queueing via the fan-out's retain. - kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - backlog_capacity, - // Fire-and-forget: the multi-shard register paths are superseded - // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up - // protocol is wired on the single-shard inline path only. - registered: None, - // Cross-shard registration: the target shard's offset is - // owned by its own thread — the arm reads it at drain. - push_offset: None, - // No snapshot body was captured on this shard's thread — - // the arm's drain-time offset is the correct cut. - cut: None, - }, - )); - let _ = prod.try_push(msg); - } - - // Spawn sender task: drains channel -> writes to TCP socket - let wh = Arc::clone(&write_half); - tokio::spawn(async move { - while let Ok(data) = rx.recv_async().await { - let mut guard = wh.lock().await; - if guard.write_all(&data).await.is_err() { - info!("Replica sender shard {}: socket closed", shard_id); - break; - } - } - }); - } - - // Register replica in ReplicationState - let replica_info = ReplicaInfo { - id: replica_id, - addr, - ack_offsets, - shard_txs, - last_ack_time: std::sync::atomic::AtomicU64::new( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - ), - }; - if let Ok(mut rs) = repl_state.write() { - rs.replicas.push(replica_info); - } - - info!( - "Master: replica {} registered across {} shards", - replica_id, num_shards - ); - Ok(()) -} - -/// Monoio variant of replica registration. -/// -/// Uses `Rc>>` with take/put-back pattern -/// for ownership I/O writes. Single-threaded cooperative scheduling ensures only -/// one sender task runs at a time. -#[cfg(feature = "runtime-monoio")] -async fn register_replica_with_shards( - addr: std::net::SocketAddr, - stream: monoio::net::TcpStream, - repl_state: Arc>, - shard_producers: &mut Vec>, - num_shards: usize, -) -> anyhow::Result<()> { - use monoio::io::AsyncWriteRentExt; - use ringbuf::traits::Producer; - use std::sync::atomic::Ordering; - - static NEXT_REPLICA_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); - let replica_id = NEXT_REPLICA_ID.fetch_add(1, Ordering::Relaxed); - - // Share the stream across per-shard sender tasks. - // monoio's write_all takes &mut self + owned buffer, so RefCell suffices. - // Single-threaded cooperative scheduling ensures no concurrent borrows. - let shared_stream: Rc> = Rc::new(RefCell::new(stream)); - - // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. - let backlog_capacity = repl_state - .read() - .map(|g| g.backlog_capacity) - .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); - - let channel_capacity = 1024; - let mut shard_txs = Vec::with_capacity(num_shards); - let mut ack_offsets = Vec::with_capacity(num_shards); - - for shard_id in 0..num_shards { - let (tx, rx) = crate::runtime::channel::mpsc_bounded::(channel_capacity); - shard_txs.push(tx.clone()); - ack_offsets.push(std::sync::atomic::AtomicU64::new(0)); - - // Send RegisterReplica to the shard's SPSC - if let Some(prod) = shard_producers.get_mut(shard_id) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica(Box::new( - crate::shard::dispatch::RegisterReplicaPayload { - replica_id, - tx, - // Legacy multi-shard drain loops do not poll the kick flag - // (superseded by the R2 redesign); overflow still stops - // queueing via the fan-out's retain. - kicked: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), - backlog_capacity, - // Fire-and-forget: the multi-shard register paths are superseded - // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up - // protocol is wired on the single-shard inline path only. - registered: None, - // Cross-shard registration: the target shard's offset is - // owned by its own thread — the arm reads it at drain. - push_offset: None, - // No snapshot body was captured on this shard's thread — - // the arm's drain-time offset is the correct cut. - cut: None, - }, - )); - let _ = prod.try_push(msg); - } - - // Spawn sender task: drains channel -> writes to TCP socket - // monoio write_all takes &mut self, so we borrow_mut() across the await. - // This is safe because monoio is single-threaded and cooperative — - // only one sender task runs at a time, so no concurrent borrows occur. - let wh = Rc::clone(&shared_stream); - #[allow(clippy::await_holding_refcell_ref)] - monoio::spawn(async move { - while let Ok(data) = rx.recv_async().await { - let data_vec = data.to_vec(); - let (wr, _) = wh.borrow_mut().write_all(data_vec).await; - if wr.is_err() { - info!("Replica sender shard {}: socket closed", shard_id); - break; - } - } - }); - } - - // Register replica in ReplicationState - let replica_info = ReplicaInfo { - id: replica_id, - addr, - ack_offsets, - shard_txs, - last_ack_time: std::sync::atomic::AtomicU64::new( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - ), - }; - if let Ok(mut rs) = repl_state.write() { - rs.replicas.push(replica_info); - } - - info!( - "Master: replica {} registered across {} shards", - replica_id, num_shards - ); - Ok(()) -} - /// Inline single-shard PSYNC handler: snapshots the local shard's databases /// directly (no SnapshotBegin SPSC self-send), sends `+FULLRESYNC` followed by /// the RDB, then registers the replica for live streaming. @@ -634,9 +69,7 @@ pub async fn handle_psync_inline_single_shard( // same synchronous stretch as the RDB capture (see below) so no write can // slip between the two. let (repl_id, repl_id2, backlog_slot) = { - let rs = repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + let rs = repl_state.read(); let slot = rs .per_shard_backlogs .first() @@ -686,24 +119,21 @@ pub async fn handle_psync_inline_single_shard( // Clone — its internal DashTable + FT/graph indices are large). let mut rdb_buf: Vec = Vec::new(); let snapshot_offset = { - let off = repl_state - .read() - .map(|g| { - // HIGH-2 (task #22): reset the stream's db context in - // the SAME synchronous stretch as the snapshot capture - // — every byte at offset ≥ snapshot_offset then starts - // from "db unknown", so the first post-snapshot write - // re-emits `SELECT ` and this replica's drain - // (which starts at db 0 after loading the RDB) can - // never bind a write to the wrong db. Redis's - // `slaveseldb = -1` idiom. Redundant re-SELECTs for - // already-attached replicas are idempotent. - if let Some(slot) = g.stream_db.first() { - slot.store(-1, std::sync::atomic::Ordering::Relaxed); - } - g.total_offset() - }) - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + let g = repl_state.read(); + // HIGH-2 (task #22): reset the stream's db context in + // the SAME synchronous stretch as the snapshot capture + // — every byte at offset ≥ snapshot_offset then starts + // from "db unknown", so the first post-snapshot write + // re-emits `SELECT ` and this replica's drain + // (which starts at db 0 after loading the RDB) can + // never bind a write to the wrong db. Redis's + // `slaveseldb = -1` idiom. Redundant re-SELECTs for + // already-attached replicas are idempotent. + if let Some(slot) = g.stream_db.first() { + slot.store(-1, std::sync::atomic::Ordering::Relaxed); + } + let off = g.total_offset(); + drop(g); // Shard 0 is this thread's shard — use the thread-local slice. crate::shard::slice::with_shard(|s| { let refs: Vec<&crate::storage::Database> = s.databases.iter().collect(); @@ -874,9 +304,7 @@ pub async fn handle_psync_inline_multi_shard( use ringbuf::traits::Producer; let (repl_id, backlog_capacity) = { - let rs = repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + let rs = repl_state.read(); (rs.repl_id.clone(), rs.backlog_capacity) }; @@ -1234,10 +662,7 @@ fn push_register_replica_inline( let kicked = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); // `--repl-backlog-size`, carried in RegisterReplica for the lazy fallback-init. - let backlog_capacity = repl_state - .read() - .map(|g| g.backlog_capacity) - .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); + let backlog_capacity = repl_state.read().backlog_capacity; // The inline PSYNC task runs ON the owning shard's thread; the SPSC mesh // has no self-loop (N·(N−1) skip-self — at shards=1 the producer Vec is // EMPTY), so registration goes through the thread-local self queue the @@ -1253,9 +678,7 @@ fn push_register_replica_inline( // offset keeps catch-up and live delivery disjoint for every interleave // (see `RegisterReplica::push_offset`). let (push_offset, push_shard_offset) = { - let g = repl_state - .read() - .map_err(|_| anyhow::anyhow!("replication state lock poisoned"))?; + let g = repl_state.read(); // Master-axis offset for the catch-up reply protocol, PER-SHARD-axis // offset for the fan-out cut. This path only runs at shards=1 // (multi-shard PSYNC routes through `handle_psync_inline_multi_shard`), @@ -1321,9 +744,7 @@ async fn drain_replica_inline_single_shard( .as_secs(), ), }; - if let Ok(mut rs) = repl_state.write() { - rs.replicas.push(replica_info); - } + repl_state.write().replicas.push(replica_info); // R1 (task #19): the hijacked PSYNC socket is full-duplex — the replica // sends `REPLCONF ACK ` back on it (1s cadence). Split the stream @@ -1378,9 +799,7 @@ async fn drain_replica_inline_single_shard( drop(ack_reader); // Remove from ReplicationState; the event loop will drop its replica_txs // entry on the next failed send via its own UnregisterReplica path. - if let Ok(mut rs) = repl_state.write() { - rs.replicas.retain(|r| r.id != replica_id); - } + repl_state.write().replicas.retain(|r| r.id != replica_id); // Same-thread → self queue (no self-SPSC exists; see push_register_replica_inline). crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::UnregisterReplica { replica_id, @@ -1425,21 +844,20 @@ async fn ack_read_loop( } }; for offset in acks { - if let Ok(rs) = repl_state.read() { - if let Some(info) = rs.replicas.iter().find(|r| r.id == replica_id) { - // fetch_max: ACKs can only move forward — a reordered or - // duplicate ACK never regresses the recorded offset. - if let Some(slot) = info.ack_offsets.first() { - slot.fetch_max(offset, Ordering::Relaxed); - } - info.last_ack_time.store( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - Ordering::Relaxed, - ); + let rs = repl_state.read(); + if let Some(info) = rs.replicas.iter().find(|r| r.id == replica_id) { + // fetch_max: ACKs can only move forward — a reordered or + // duplicate ACK never regresses the recorded offset. + if let Some(slot) = info.ack_offsets.first() { + slot.fetch_max(offset, Ordering::Relaxed); } + info.last_ack_time.store( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + Ordering::Relaxed, + ); } } } @@ -1498,7 +916,7 @@ pub async fn wait_for_replicas( repl_state: &Arc>, ) -> usize { let target_offset = { - let rs = repl_state.read().unwrap(); + let rs = repl_state.read(); rs.total_offset() }; @@ -1506,7 +924,7 @@ pub async fn wait_for_replicas( loop { let acked_count = { - let rs = repl_state.read().unwrap(); + let rs = repl_state.read(); rs.replicas .iter() .filter(|r| { @@ -1535,6 +953,7 @@ pub async fn wait_for_replicas( #[cfg(test)] mod tests { + #[cfg(feature = "runtime-tokio")] use super::*; #[cfg(feature = "runtime-tokio")] diff --git a/src/replication/reason_del.rs b/src/replication/reason_del.rs index be0760ac..82a7ebd0 100644 --- a/src/replication/reason_del.rs +++ b/src/replication/reason_del.rs @@ -139,7 +139,7 @@ pub(crate) fn record_reason_del( #[cfg(feature = "runtime-monoio")] #[allow(clippy::too_many_arguments)] pub(crate) fn record_reason_del_conn( - repl_state: &Option>>, + repl_state: &Option>>, shard_id: usize, num_shards: usize, aof_pool: Option<&std::sync::Arc>, @@ -177,7 +177,7 @@ pub(crate) fn record_reason_del_conn( /// PSYNC and the shard self-msg relay this rides on are monoio-only. #[cfg(feature = "runtime-monoio")] pub(crate) fn record_effect_write( - repl_state: &Option>>, + repl_state: &Option>>, shard_id: usize, num_shards: usize, aof_pool: Option<&std::sync::Arc>, @@ -219,7 +219,7 @@ fn conn_has_work(aof_pool: Option<&std::sync::Arc>) -> bool { /// mechanics for an arbitrary pre-serialized record instead of only `DEL`. #[cfg(feature = "runtime-monoio")] fn record_bytes_conn( - repl_state: &Option>>, + repl_state: &Option>>, shard_id: usize, num_shards: usize, aof_pool: Option<&std::sync::Arc>, @@ -246,7 +246,7 @@ fn record_bytes_conn( /// dispatch fast path only threads the individual fields it needs). #[cfg(feature = "runtime-monoio")] fn push_record_db( - repl_state: &Option>>, + repl_state: &Option>>, shard_id: usize, num_shards: usize, db: usize, @@ -261,15 +261,13 @@ fn push_record_db( return; } let needs_select = repl_state.as_ref().is_some_and(|rs| { - rs.read().is_ok_and(|g| { - g.stream_db.get(shard_id).is_some_and(|slot| { - if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 { - slot.store(db as i64, std::sync::atomic::Ordering::Relaxed); - true - } else { - false - } - }) + rs.read().stream_db.get(shard_id).is_some_and(|slot| { + if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 { + slot.store(db as i64, std::sync::atomic::Ordering::Relaxed); + true + } else { + false + } }) }); if needs_select { @@ -286,20 +284,19 @@ fn push_record_db( /// mirrors `handler_monoio::ft::record_local_write` exactly. #[cfg(feature = "runtime-monoio")] fn push_record( - repl_state: &Option>>, + repl_state: &Option>>, shard_id: usize, bytes: Bytes, ) { let mut end_offset = u64::MAX; if let Some(rs) = repl_state.as_ref() { - if let Ok(g) = rs.read() { - if let Some(slot) = g.per_shard_backlogs.get(shard_id) { - if let Some(backlog) = slot.lock().as_mut() { - backlog.append(&bytes); - } + let g = rs.read(); + if let Some(slot) = g.per_shard_backlogs.get(shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&bytes); } - end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); } + end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); } crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes, diff --git a/src/replication/replica.rs b/src/replication/replica.rs index cd80b8cb..e203601f 100644 --- a/src/replication/replica.rs +++ b/src/replication/replica.rs @@ -6,8 +6,9 @@ #![allow(unused_imports)] use bytes::{Bytes, BytesMut}; +use parking_lot::RwLock; +use std::sync::Arc; use std::sync::atomic::Ordering; -use std::sync::{Arc, RwLock}; use std::time::Duration; #[cfg(feature = "runtime-tokio")] use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -132,7 +133,8 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { } // Update handshake state to Disconnected in ReplicationState - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Disconnected; } @@ -170,7 +172,8 @@ async fn run_handshake_and_stream( } // Step 1: PING - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::PingSent; } @@ -200,10 +203,7 @@ async fn run_handshake_and_stream( // Step 4: PSYNC let (repl_id, offset) = { - let rs = cfg - .repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + let rs = cfg.repl_state.read(); let offset = rs.master_repl_offset.load(Ordering::Relaxed); let id = if offset == 0 { "?".to_string() @@ -217,7 +217,8 @@ async fn run_handshake_and_stream( }; (id, off_str) }; - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::PsyncPending; } @@ -240,7 +241,8 @@ async fn run_handshake_and_stream( .and_then(|s| s.trim().parse().ok()) .unwrap_or(0); // Update local replication state with master's repl_id - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); rs.repl_id = master_id; rs.master_repl_offset .store(master_offset, Ordering::Relaxed); @@ -290,7 +292,8 @@ async fn run_handshake_and_stream( } // Enter streaming mode - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Streaming; } @@ -302,7 +305,8 @@ async fn run_handshake_and_stream( stream_commands(stream, cfg).await?; } else if response.starts_with(b"+CONTINUE") { // Partial resync: stream from current offset - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Streaming; } @@ -340,10 +344,7 @@ async fn stream_commands(stream: TcpStream, cfg: &ReplicaTaskConfig) -> anyhow:: if stop.load(Ordering::Relaxed) { break; } - let offset = match repl_state.read() { - Ok(rs) => rs.master_repl_offset.load(Ordering::Relaxed), - Err(_) => break, - }; + let offset = repl_state.read().master_repl_offset.load(Ordering::Relaxed); if wr.write_all(&encode_replconf_ack(offset)).await.is_err() { break; // socket dead — the read loop is exiting too } @@ -407,7 +408,8 @@ async fn stream_commands_read_loop( } } if outcome.consumed > 0 { - if let Ok(rs) = cfg.repl_state.read() { + { + let rs = cfg.repl_state.read(); rs.master_repl_offset .fetch_add(outcome.consumed as u64, Ordering::Relaxed); } @@ -481,7 +483,8 @@ pub async fn run_replica_task(cfg: ReplicaTaskConfig) { return; } - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Disconnected; } @@ -522,7 +525,8 @@ async fn run_handshake_and_stream( } // Step 1: PING - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::PingSent; } @@ -552,10 +556,7 @@ async fn run_handshake_and_stream( // Step 4: PSYNC let (repl_id, offset) = { - let rs = cfg - .repl_state - .read() - .map_err(|_| anyhow::anyhow!("lock poisoned"))?; + let rs = cfg.repl_state.read(); let offset = rs.master_repl_offset.load(Ordering::Relaxed); let id = if offset == 0 { "?".to_string() @@ -569,7 +570,8 @@ async fn run_handshake_and_stream( }; (id, off_str) }; - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::PsyncPending; } @@ -590,7 +592,8 @@ async fn run_handshake_and_stream( .ok() .and_then(|s| s.trim().parse().ok()) .unwrap_or(0); - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); rs.repl_id = master_id; rs.master_repl_offset .store(master_offset, Ordering::Relaxed); @@ -635,7 +638,8 @@ async fn run_handshake_and_stream( } } - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Streaming; } @@ -646,7 +650,8 @@ async fn run_handshake_and_stream( cfg.stream_db.store(0, Ordering::Relaxed); stream_commands(stream, cfg).await?; } else if response.starts_with(b"+CONTINUE") { - if let Ok(mut rs) = cfg.repl_state.write() { + { + let mut rs = cfg.repl_state.write(); if let ReplicationRole::Replica { ref mut state, .. } = rs.role { *state = ReplicaHandshakeState::Streaming; } @@ -686,10 +691,7 @@ async fn stream_commands( if stop.get() { break; } - let offset = match repl_state.read() { - Ok(rs) => rs.master_repl_offset.load(Ordering::Relaxed), - Err(_) => break, - }; + let offset = repl_state.read().master_repl_offset.load(Ordering::Relaxed); let (res, _) = wr.write_all(encode_replconf_ack(offset)).await; if res.is_err() { break; // socket dead — the read loop is exiting too @@ -762,7 +764,8 @@ async fn stream_commands_read_loop( } } if outcome.consumed > 0 { - if let Ok(rs) = cfg.repl_state.read() { + { + let rs = cfg.repl_state.read(); rs.master_repl_offset .fetch_add(outcome.consumed as u64, Ordering::Relaxed); } diff --git a/src/replication/state.rs b/src/replication/state.rs index 0edf0d78..349bb2ce 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -122,13 +122,33 @@ impl ReplicationState { /// subsequent writes on the shard's event loop start being captured for /// partial resync. Capacity comes from `self.backlog_capacity` /// (`--repl-backlog-size`, default 1 MiB per shard). + /// + /// Task #70: allocation is deliberately split from [`mark_fanout_active`] + /// — a bare `REPLCONF` (sent by any prober, or the first two steps of a + /// handshake that never reaches `PSYNC`) used to flip the process-global + /// `FANOUT_HINT` here, and the hint is never cleared, so a single failed + /// handshake permanently taxed every future write with the replication + /// serialize+SPSC round trip. Backlog allocation itself stays load- + /// bearing on `REPLCONF` (a real handshake's `REPLCONF` arrives before + /// `PSYNC`, so deferring allocation to `PSYNC` would buffer zero bytes + /// during the handshake window); only the hint activation moved to + /// `try_handle_psync` (dispatch.rs), which runs on an actual `PSYNC` + /// arrival — never on a `REPLCONF`-only probe. + /// + /// ⚠ Correctness invariant (task #70 follow-up): allocating here seeds + /// the backlog at the shard's offset AT ALLOCATION TIME, but the gap + /// between a bare `REPLCONF` and an actual `PSYNC`/replica registration + /// can be arbitrarily long (or never close, for a probe). During that + /// gap `FANOUT_HINT` is false, so local writes take the bare-`issue_lsn` + /// branch and advance the shard offset with NO backlog append — the + /// allocated backlog silently skews stale. Backlog byte positions are + /// therefore only trustworthy from the LAST activation-time + /// [`Self::realign_backlog`] call onward, never from allocation alone. + /// Every activation site (`try_handle_psync`, and the per-shard + /// `RegisterReplica`/`PrepareReplicaSync` arms in `spsc_handler.rs`) + /// MUST call `realign_backlog` before capturing any snapshot/cut offset + /// from this backlog. pub fn ensure_backlogs_allocated(&self) { - // Hint FIRST: any write racing this allocation that still sees the - // hint as false advances the offset without a backlog append, and the - // seed below (reading the offset AFTER that advance) re-aligns. At - // shards=1 both run on the same shard thread, so there is no race at - // all. (Multi-shard masters ride the R2 redesign.) - mark_fanout_active(); for (shard_id, slot) in self.per_shard_backlogs.iter().enumerate() { let mut guard = slot.lock(); if guard.is_none() { @@ -147,6 +167,45 @@ impl ReplicationState { } } + /// Realign an already-allocated backlog for `shard_id` to the shard's + /// CURRENT real offset, discarding any bytes/positions that skewed away + /// from the offset counter during the REPLCONF-allocated / + /// FANOUT_HINT-false window (see [`Self::ensure_backlogs_allocated`]). + /// No-op if the shard has no backlog allocated yet, or if the backlog + /// is already aligned (`ReplicationBacklog::realign_to` is itself a + /// no-op in that case). + /// + /// MUST be called at every activation site — an actual PSYNC arrival + /// (`try_handle_psync`) or a shard's own `RegisterReplica` / + /// `PrepareReplicaSync` arm (`spsc_handler.rs`) — BEFORE that call site + /// captures any snapshot/cut/push offset from this backlog. A skewed + /// buffer's contents are already useless (they cover positions the real + /// counter never confirmed), so resetting is correct: any partial-resync + /// request for an offset below the new `start_offset` falls through + /// `bytes_from` returning `None`, and the caller's existing fail-safe — + /// full resync — kicks in. Never silently serves wrong bytes. + /// + /// Race-free ONLY when called from the shard thread that owns + /// `shard_id`'s own offset advances — true for every call site above + /// (the inline single-shard PSYNC leg at shards=1 runs on that shard's + /// event-loop thread; the SPSC arms run there by construction), so the + /// offset read below and the realign happen back-to-back with nothing + /// else able to interleave on that shard's own append/advance sequence. + pub fn realign_backlog(&self, shard_id: usize) { + let Some(slot) = self.per_shard_backlogs.get(shard_id) else { + return; + }; + let mut guard = slot.lock(); + if let Some(backlog) = guard.as_mut() { + let offset = self + .shard_offsets + .get(shard_id) + .map(|o| o.load(Ordering::Relaxed)) + .unwrap_or(0); + backlog.realign_to(offset); + } + } + /// Total resident bytes across all per-shard replication backlogs. /// Returns 0 if no backlogs have been allocated (lazy init). /// O(num_shards) -- one lock acquire per shard (uncontended on metrics scrape). @@ -387,9 +446,15 @@ pub fn save_replication_state( /// Write hot paths gate their replication-fanout serialization on this ONE /// Relaxed load instead of taking `repl_state.read()` per command (the same /// S3.5a rationale that gave READONLY its `is_replica_mirror` AtomicBool). -/// Set by `ensure_backlogs_allocated` (REPLCONF/PSYNC arrival) and never -/// cleared — once a replica has attached, the master keeps feeding the -/// backlog so partial resync stays possible, exactly like Redis. +/// +/// Set by an actual `PSYNC` arrival (`try_handle_psync` in +/// `handler_monoio::dispatch`) or multi-shard replica registration — NOT by +/// a bare `REPLCONF` (task #70: `REPLCONF` only allocates the backlog via +/// [`ReplicationState::ensure_backlogs_allocated`], which does not touch this +/// hint, so a prober or a handshake that never reaches `PSYNC` cannot flip it +/// on). Never cleared once set — once a replica has genuinely begun +/// attaching, the master keeps feeding the backlog so partial resync stays +/// possible, exactly like Redis. static FANOUT_HINT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); /// Mark the fanout hint (idempotent). See [`FANOUT_HINT`]. @@ -427,15 +492,14 @@ pub fn record_local_write_global(shard_id: usize, bytes: Bytes) { let Some(repl_state_arc) = crate::admin::metrics_setup::get_global_repl_state_arc() else { return; }; - let mut end_offset = u64::MAX; - if let Ok(g) = repl_state_arc.read() { - if let Some(slot) = g.per_shard_backlogs.get(shard_id) { - if let Some(backlog) = slot.lock().as_mut() { - backlog.append(&bytes); - } + let g = repl_state_arc.read(); + if let Some(slot) = g.per_shard_backlogs.get(shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&bytes); } - end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); } + let end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); + drop(g); crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes, end_offset, @@ -453,16 +517,18 @@ pub fn record_local_write_db_global(shard_id: usize, db: usize, bytes: Bytes) { let Some(repl_state_arc) = crate::admin::metrics_setup::get_global_repl_state_arc() else { return; }; - let needs_select = repl_state_arc.read().is_ok_and(|g| { - g.stream_db.get(shard_id).is_some_and(|slot| { + let needs_select = repl_state_arc + .read() + .stream_db + .get(shard_id) + .is_some_and(|slot| { if slot.load(Ordering::Relaxed) != db as i64 { slot.store(db as i64, Ordering::Relaxed); true } else { false } - }) - }); + }); if needs_select { record_local_write_global( shard_id, @@ -691,6 +757,194 @@ mod tests { assert_eq!(backlog.end_offset(), (MIN_REPL_BACKLOG_SIZE + 100) as u64); } + /// Task #70 RED/GREEN: `ensure_backlogs_allocated` — the side effect of a + /// bare REPLCONF (`try_handle_replconf`, dispatch.rs) — must NOT flip the + /// sticky, process-global `FANOUT_HINT`. Only an actual PSYNC arrival + /// (`try_handle_psync`) or replica registration (`RegisterReplica`/ + /// `PrepareReplicaSync` in `spsc_handler.rs`) may. + /// + /// RED on the pre-fix code: the old `ensure_backlogs_allocated` called + /// `mark_fanout_active()` unconditionally as its first line, so this + /// delta assertion (`before == after`) fails — `after` flips to `true` + /// regardless of `before` — proving a probe/failed handshake that only + /// ever sends REPLCONF (never PSYNC) permanently taxed every future + /// write. GREEN on the fix: `ensure_backlogs_allocated` only allocates + /// the backlog (still load-bearing, asserted below), the hint is + /// untouched. + /// + /// `FANOUT_HINT` is a process-global `static` shared by every test in + /// this binary; the delta form (`before == after`) is deliberately + /// robust to whatever the ambient value already is — no test in this + /// codebase's unit-test-reachable surface calls `mark_fanout_active` + /// outside a real `ConnectionContext`/`RegisterReplica` path, so `before` + /// is `false` in practice, but the assertion does not depend on that. + #[test] + fn test_ensure_backlogs_allocated_does_not_activate_fanout_hint() { + let before = fanout_hint_active(); + let state = ReplicationState::new(2, generate_repl_id(), ZEROED_ID.to_string()); + state.ensure_backlogs_allocated(); + let after = fanout_hint_active(); + assert_eq!( + before, after, + "ensure_backlogs_allocated (bare-REPLCONF path) must not change \ + the sticky process-global fanout hint — task #70" + ); + // Allocation itself is unchanged, still load-bearing: writes between + // REPLCONF and PSYNC are captured once PSYNC actually flips the hint. + assert!( + state.per_shard_backlogs[0].lock().is_some(), + "backlog allocation must still happen on REPLCONF" + ); + assert!(state.per_shard_backlogs[1].lock().is_some()); + } + + /// Task #70 companion: the actual activation path (`mark_fanout_active`, + /// called from `try_handle_psync` on a genuine PSYNC arrival and from the + /// `RegisterReplica`/`PrepareReplicaSync` shard-message arms) still sets + /// the hint — monotonic, so this holds regardless of the ambient value. + #[test] + fn test_mark_fanout_active_sets_hint() { + mark_fanout_active(); + assert!( + fanout_hint_active(), + "mark_fanout_active (actual PSYNC/replica registration) must set the hint" + ); + } + + /// Task #70 correctness follow-up (orchestrator-caught regression): + /// `ensure_backlogs_allocated` (bare REPLCONF) seeds the backlog at the + /// shard's offset AT THAT INSTANT. If the hint stays false afterward + /// (no PSYNC yet — exactly the window this task's fix widened), local + /// writes advance the shard offset via the bare `issue_lsn` path with NO + /// backlog append, so the backlog silently skews stale relative to the + /// real offset counter. This test proves the skew is real (asserting it + /// BEFORE calling the fix), then proves `realign_backlog` — called at + /// every activation site — repairs it. + /// + /// RED without the fix: if a caller captured a snapshot/cut offset from + /// the backlog at this point (skipping `realign_backlog`), it would read + /// `backlog.end_offset() == 0` while the shard's real offset is `300` — + /// a 300-byte skew. `bytes_from(300)` against the unrealigned backlog + /// returns `None` (300 is treated as a future offset past a backlog + /// whose `end_offset` is still 0), which silently drops partial-resync + /// catch-up instead of correctly triggering a full resync at the RIGHT + /// offset — the corruption the orchestrator flagged. + /// GREEN with the fix: after `realign_backlog`, `end_offset` matches the + /// real offset exactly and `bytes_from` at that offset returns the + /// (empty) live tail, not `None`. + #[test] + fn test_realign_backlog_fixes_skew_after_hint_false_window() { + let state = ReplicationState::new(1, generate_repl_id(), ZEROED_ID.to_string()); + + // 1. Bare REPLCONF: allocates + seeds the backlog at offset 0. + state.ensure_backlogs_allocated(); + assert_eq!( + state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .end_offset(), + 0 + ); + + // 2. The FANOUT_HINT-false window: local writes advance the shard's + // real offset via the bare-LSN path (no backlog append — this is + // exactly what `record_local_write` skips while + // `replication_fanout_active` is false). + state.issue_lsn(0, 100); + state.issue_lsn(0, 150); + state.issue_lsn(0, 50); + let real_offset = state.shard_offset(0); + assert_eq!(real_offset, 300); + + // Prove the skew actually exists before the fix runs: the backlog + // is still stuck at its allocation-time offset while the real + // counter has moved on. + let skewed_end = state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .end_offset(); + assert_eq!(skewed_end, 0, "backlog must still be stale pre-realign"); + assert_ne!( + skewed_end, real_offset, + "skew must exist: this is the bug the orchestrator caught" + ); + // A caller that (incorrectly) trusted the skewed backlog now would + // silently drop catch-up instead of full-resyncing: bytes_from at + // the real offset returns None because the backlog doesn't know + // any offset that high exists yet. + assert_eq!( + state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .bytes_from(real_offset), + None, + "unrealigned backlog treats the real offset as an unreachable future position" + ); + + // 3. Activation (PSYNC arrival / RegisterReplica arm): realign BEFORE + // any snapshot/cut offset is captured. + state.realign_backlog(0); + + let realigned_end = state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .end_offset(); + assert_eq!( + realigned_end, real_offset, + "realign_backlog must bring the backlog's end_offset back in sync \ + with the shard's real offset counter — task #70 correctness fix" + ); + assert_eq!( + state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .bytes_from(real_offset), + Some(vec![]), + "post-realign, the real offset is a valid (empty) live-tail position" + ); + } + + /// Integration-shaped companion: after realign, a record appended via + /// the normal write path lands at exactly the pre-append real offset — + /// end-to-end proof that catch-up reads from the correct byte range + /// once activation has realigned the backlog. + #[test] + fn test_realign_backlog_then_append_reads_back_exact_record() { + let state = ReplicationState::new(1, generate_repl_id(), ZEROED_ID.to_string()); + state.ensure_backlogs_allocated(); + state.issue_lsn(0, 300); // hint-false window skew, as above. + state.realign_backlog(0); // activation. + + let pre_append_offset = state.shard_offset(0); + assert_eq!(pre_append_offset, 300); + + // Simulate the write path's post-activation record append + // (`record_local_write`'s backlog.append + increment_shard_offset + // pairing) directly against the now-realigned backlog. + let record = b"*1\r\n$4\r\nPING\r\n"; + { + let mut guard = state.per_shard_backlogs[0].lock(); + guard.as_mut().unwrap().append(record); + } + state.increment_shard_offset(0, record.len() as u64); + + let bytes = state.per_shard_backlogs[0] + .lock() + .as_ref() + .unwrap() + .bytes_from(pre_append_offset) + .expect("pre_append_offset must still be live after realign+append"); + assert_eq!( + bytes, record, + "catch-up from the pre-append offset must return exactly that record's bytes" + ); + } + #[test] fn test_is_replica_mirror_default_false() { let state = ReplicationState::new(1, generate_repl_id(), ZEROED_ID.to_string()); diff --git a/src/scripting/bridge.rs b/src/scripting/bridge.rs index 135cec46..927f3d39 100644 --- a/src/scripting/bridge.rs +++ b/src/scripting/bridge.rs @@ -89,7 +89,7 @@ struct LuaEvictionInner { #[cfg_attr(not(feature = "runtime-monoio"), allow(dead_code))] num_shards: usize, #[cfg_attr(not(feature = "runtime-monoio"), allow(dead_code))] - repl_state: Option>>, + repl_state: Option>>, #[cfg_attr(not(feature = "runtime-monoio"), allow(dead_code))] aof_pool: Option>, /// Task #38: lock-free snapshot of `ReplicationState::is_replica_mirror`, @@ -128,7 +128,7 @@ impl LuaEvictionCtx { spill_file_id: Rc>, disk_offload_dir: Option, num_shards: usize, - repl_state: Option>>, + repl_state: Option>>, aof_pool: Option>, ) -> Self { // Task #38: snapshot the lock-free mirror the same way @@ -137,7 +137,7 @@ impl LuaEvictionCtx { // `ReplicationState::set_role()` writing the same `AtomicBool`. let is_replica_mirror = repl_state .as_ref() - .and_then(|rs| rs.read().ok().map(|guard| guard.is_replica_mirror.clone())); + .map(|rs| rs.read().is_replica_mirror.clone()); LuaEvictionCtx(Some(LuaEvictionInner { shard_databases, runtime_config, @@ -610,7 +610,7 @@ mod tests { let (shard_databases, _inits) = ShardDatabases::new(vec![vec![Database::new()]]); let runtime_config = Arc::new(parking_lot::RwLock::new(make_config(0, "noeviction"))); - let repl_state = Arc::new(std::sync::RwLock::new(ReplicationState::new( + let repl_state = Arc::new(parking_lot::RwLock::new(ReplicationState::new( 1, "a".repeat(40), "0".repeat(40), @@ -633,23 +633,17 @@ mod tests { "fresh ReplicationState defaults to Master" ); - repl_state - .write() - .unwrap() - .set_role(ReplicationRole::Replica { - host: "127.0.0.1".to_string(), - port: 6379, - state: ReplicaHandshakeState::PingPending, - }); + repl_state.write().set_role(ReplicationRole::Replica { + host: "127.0.0.1".to_string(), + port: 6379, + state: ReplicaHandshakeState::PingPending, + }); assert!( ctx.is_replica(), "is_replica() must observe a role flip that happened after ctx construction" ); - repl_state - .write() - .unwrap() - .set_role(ReplicationRole::Master); + repl_state.write().set_role(ReplicationRole::Master); assert!( !ctx.is_replica(), "REPLICAOF NO ONE must flip is_replica() back to false" diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs index 482de9bb..d0383f34 100644 --- a/src/server/conn/blocking.rs +++ b/src/server/conn/blocking.rs @@ -1147,7 +1147,7 @@ pub(crate) fn try_inline_dispatch( selected_db: usize, aof_pool: &Option>, repl_state: &Option< - std::sync::Arc>, + std::sync::Arc>, >, now_ms: u64, num_shards: usize, @@ -1512,7 +1512,7 @@ pub(crate) fn try_inline_dispatch_loop( selected_db: usize, aof_pool: &Option>, repl_state: &Option< - std::sync::Arc>, + std::sync::Arc>, >, now_ms: u64, num_shards: usize, diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index 8cd0c4a8..ae195618 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -34,6 +34,9 @@ use crate::workspace::WorkspaceId; use super::affinity::{AffinityTracker, MigratedConnectionState}; /// Type alias for std::sync::RwLock to distinguish from parking_lot::RwLock. +/// `ReplicationState` no longer uses this (task #70 — migrated to +/// `parking_lot::RwLock`, see `repl_state` below); it remains in use for +/// `acl_table` and `cluster_state`, which are out of scope for that migration. pub(crate) type StdRwLock = std::sync::RwLock; /// Immutable context shared across all connections on a shard. @@ -55,7 +58,7 @@ pub(crate) struct ConnectionContext { /// PerShard until per-shard rewrite ships (step 6 of the RFC). pub aof_pool: Option>, pub tracking_table: std::sync::Arc>, - pub repl_state: Option>>, + pub repl_state: Option>>, /// Lock-free mirror of `repl_state.role == Replica { .. }`. /// Populated once in `new()` (from `repl_state.read()`), kept in sync /// thereafter by `ReplicationState::set_role()`. Read on every command @@ -104,7 +107,7 @@ impl ConnectionContext { requirepass: Option, aof_pool: Option>, tracking_table: std::sync::Arc>, - repl_state: Option>>, + repl_state: Option>>, cluster_state: Option>>, lua: Rc, script_cache: Rc>, @@ -134,7 +137,7 @@ impl ConnectionContext { // ReplicationState::set_role() updates the same AtomicBool thereafter. let is_replica_mirror = repl_state .as_ref() - .and_then(|rs| rs.read().ok().map(|guard| guard.is_replica_mirror.clone())); + .map(|rs| rs.read().is_replica_mirror.clone()); Self { shard_databases, shard_id, diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index fffc58df..70ec47a8 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -455,14 +455,13 @@ pub(super) fn try_handle_replicaof( if let Some(ref rs) = ctx.repl_state { match action { ReplicaofAction::StartReplication { host, port } => { - if let Ok(mut rs_guard) = rs.write() { - rs_guard.set_role(crate::replication::state::ReplicationRole::Replica { + rs.write() + .set_role(crate::replication::state::ReplicationRole::Replica { host: host.clone(), port, state: crate::replication::handshake::ReplicaHandshakeState::PingPending, }); - } let rs_clone = Arc::clone(rs); // Bump the task generation FIRST: any previously spawned // replica task (old REPLICAOF target) sees itself @@ -488,11 +487,10 @@ pub(super) fn try_handle_replicaof( // left it streaming + applying forever (each NO ONE → // re-attach cycle stacked one more live applier). let _ = crate::replication::replica::bump_replica_task_epoch(); - if let Ok(mut rs_guard) = rs.write() { - rs_guard.repl_id2 = rs_guard.repl_id.clone(); - rs_guard.repl_id = generate_repl_id(); - rs_guard.set_role(crate::replication::state::ReplicationRole::Master); - } + let mut rs_guard = rs.write(); + rs_guard.repl_id2 = rs_guard.repl_id.clone(); + rs_guard.repl_id = generate_repl_id(); + rs_guard.set_role(crate::replication::state::ReplicationRole::Master); } ReplicaofAction::NoOp => {} } @@ -510,6 +508,12 @@ pub(super) fn try_handle_replicaof( /// chicken-and-egg gap where the original code only allocated on /// `RegisterReplica` (after PSYNC), causing the master to buffer zero bytes /// during the handshake window. +/// +/// Task #70: allocation-only — does NOT activate the sticky `FANOUT_HINT`. +/// A bare REPLCONF (health-checker probe, or a handshake that never sends +/// PSYNC) must not permanently tax every subsequent write with the +/// replication serialize+SPSC round trip. The hint is activated on the +/// actual PSYNC arrival instead (`try_handle_psync` above). #[inline] pub(super) fn try_handle_replconf( cmd: &[u8], @@ -521,10 +525,9 @@ pub(super) fn try_handle_replconf( return false; } if let Some(ref rs) = ctx.repl_state { - if let Ok(g) = rs.read() { - if matches!(g.role, crate::replication::state::ReplicationRole::Master) { - g.ensure_backlogs_allocated(); - } + let g = rs.read(); + if matches!(g.role, crate::replication::state::ReplicationRole::Master) { + g.ensure_backlogs_allocated(); } } responses.push(crate::command::connection::replconf(cmd_args)); @@ -577,20 +580,31 @@ pub(super) fn try_handle_psync( return None; }; { - let g = rs.read().ok(); - let is_master = g - .as_ref() - .map(|g| matches!(g.role, crate::replication::state::ReplicationRole::Master)) - .unwrap_or(false); + let g = rs.read(); + let is_master = matches!(g.role, crate::replication::state::ReplicationRole::Master); if !is_master { responses.push(Frame::Error(Bytes::from_static( b"ERR PSYNC is only valid on a master", ))); return None; } - if let Some(g) = g { - g.ensure_backlogs_allocated(); - } + g.ensure_backlogs_allocated(); + // Task #70: this is the actual-PSYNC arrival — activate the sticky + // fanout hint HERE, not on a bare REPLCONF (see + // `ReplicationState::ensure_backlogs_allocated` doc comment). A + // handshake that reaches PSYNC is a genuine replica attaching; a + // REPLCONF-only probe never reaches this line. + crate::replication::state::mark_fanout_active(); + // Correctness fix (orchestrator-caught regression on the above): + // a bare REPLCONF may have already allocated this shard's backlog + // and left it skewed stale during the FANOUT_HINT-false window (see + // `ReplicationState::realign_backlog` doc comment). Realign BEFORE + // any snapshot/cut offset is captured below — this function runs on + // the connection task, which at shards=1 (the only case this + // single-shard inline leg handles) IS the owning shard's own + // event-loop thread, so the offset read inside `realign_backlog` + // and this shard's own append/advance sequence cannot interleave. + g.realign_backlog(ctx.shard_id); } let repl_id = match &cmd_args[0] { Frame::BulkString(b) | Frame::SimpleString(b) => String::from_utf8_lossy(b).into_owned(), @@ -662,7 +676,7 @@ pub(super) async fn try_handle_info( }); let mut response_text = response_text; if let Some(ref rs) = ctx.repl_state { - if let Ok(rs_guard) = rs.try_read() { + if let Some(rs_guard) = rs.try_read() { response_text.push_str(&crate::replication::handshake::build_info_replication( &rs_guard, )); diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index ecd49b6a..fb7021e0 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -32,6 +32,15 @@ fn is_replicated_ft_def_mutation(cmd: &[u8], cmd_args: &[Frame]) -> bool { /// snapshot is captured, so an FT.* mutation racing a first-ever attach still /// fans out). Gates the serialize+SPSC round trip so servers that never /// replicate pay nothing — mirrors `wal_fanout_has_work` on the KV path. +/// +/// This is a deliberately SHORT-LIVED, independent lock from the one +/// [`record_local_write_db`]/[`record_local_write`] take: several call sites +/// reach an `.await` (e.g. the AOF `send_append_group` hop) between this gate +/// check and their next `repl_state` touch, so the gate's guard must not +/// outlive this function (task #70 — never hold a lock across `.await`). +/// Only the WRITE itself — `record_local_write_db`'s own internal chain, +/// entirely synchronous — collapses to one shared guard; see there. +#[inline] pub(super) fn replication_fanout_active(ctx: &ConnectionContext) -> bool { // Cheap first gate: one Relaxed load; false until the first replica ever // begins attaching (REPLCONF/PSYNC → ensure_backlogs_allocated). @@ -39,12 +48,11 @@ pub(super) fn replication_fanout_active(ctx: &ConnectionContext) -> bool { return false; } ctx.repl_state.as_ref().is_some_and(|rs| { - rs.read().is_ok_and(|g| { - !g.replicas.is_empty() - || g.per_shard_backlogs - .first() - .is_some_and(|slot| slot.lock().is_some()) - }) + let g = rs.read(); + !g.replicas.is_empty() + || g.per_shard_backlogs + .first() + .is_some_and(|slot| slot.lock().is_some()) }) } @@ -73,23 +81,28 @@ pub(super) fn replication_fanout_active(ctx: &ConnectionContext) -> bool { /// source of truth, and a later lazy allocation seeds the backlog at the /// current counter (`ReplicationBacklog::new_at`). /// +/// Takes the `ReplicationState` guard by reference (task #70) instead of +/// re-locking `ctx.repl_state` itself, so a caller writing both a `SELECT` +/// prefix and a payload record (see [`record_local_write_db`]) pays for +/// exactly one `.read()` acquisition, not one per record. +/// /// ⚠ Monoio shard threads only (pushes to `shard::self_msg`) — callers are /// all inside `handler_monoio`, which is `runtime-monoio`-gated. -pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { +#[inline] +pub(super) fn record_local_write( + g: &crate::replication::state::ReplicationState, + shard_id: usize, + bytes: Bytes, +) { // Per-shard offset AFTER this record — the fan-out arm compares it // against each replica's snapshot cut (`ReplicaFanout::cut`) so a record // already inside a FULLRESYNC body is never live-delivered again. - let mut end_offset = u64::MAX; - if let Some(rs) = ctx.repl_state.as_ref() { - if let Ok(g) = rs.read() { - if let Some(slot) = g.per_shard_backlogs.get(ctx.shard_id) { - if let Some(backlog) = slot.lock().as_mut() { - backlog.append(&bytes); - } - } - end_offset = g.increment_shard_offset(ctx.shard_id, bytes.len() as u64); + if let Some(slot) = g.per_shard_backlogs.get(shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&bytes); } } + let end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64); crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes, end_offset, @@ -109,7 +122,20 @@ pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { /// "stream db context == the writing connection's selected db", so a /// db-agnostic record can never silently strand a stale context for the next /// KV write. +/// +/// Task #70: takes exactly ONE `ctx.repl_state.read()` guard for the whole +/// call — shared by the `needs_select` decision and both potential +/// `record_local_write` calls (SELECT prefix + payload) — collapsing what was +/// previously up to 3 separate acquisitions (plus the caller's own +/// `replication_fanout_active` gate check) down to 1 here. `None` (no +/// `repl_state` configured) is a silent no-op, matching the prior behavior. +#[inline] pub(super) fn record_local_write_db(ctx: &ConnectionContext, db: usize, bytes: Bytes) { + let Some(rs) = ctx.repl_state.as_ref() else { + return; + }; + let g = rs.read(); + // R2 (task #20): multi-shard masters merge N shard streams onto one // replica wire, so the per-shard `stream_db` context tracking below is // meaningless there — another shard may have moved the wire's db context @@ -126,28 +152,24 @@ pub(super) fn record_local_write_db(ctx: &ConnectionContext, db: usize, bytes: B let mut combined = Vec::with_capacity(select.len() + bytes.len()); combined.extend_from_slice(&select); combined.extend_from_slice(&bytes); - record_local_write(ctx, Bytes::from(combined)); + record_local_write(&g, ctx.shard_id, Bytes::from(combined)); } else { - record_local_write(ctx, bytes); + record_local_write(&g, ctx.shard_id, bytes); } return; } - let needs_select = ctx.repl_state.as_ref().is_some_and(|rs| { - rs.read().is_ok_and(|g| { - g.stream_db.get(ctx.shard_id).is_some_and(|slot| { - if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 { - slot.store(db as i64, std::sync::atomic::Ordering::Relaxed); - true - } else { - false - } - }) - }) + let needs_select = g.stream_db.get(ctx.shard_id).is_some_and(|slot| { + if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 { + slot.store(db as i64, std::sync::atomic::Ordering::Relaxed); + true + } else { + false + } }); if needs_select { - record_local_write(ctx, Bytes::from(serialize_select(db))); + record_local_write(&g, ctx.shard_id, Bytes::from(serialize_select(db))); } - record_local_write(ctx, bytes); + record_local_write(&g, ctx.shard_id, bytes); } /// RESP-serialize `SELECT ` for the replication stream. diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 16c50cad..844b8a92 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -62,8 +62,9 @@ pub enum MonoioHandlerResult { target_shard: usize, }, /// PSYNC arrived on this connection. Caller must hand the underlying - /// `monoio::net::TcpStream` to `crate::replication::master::handle_psync_on_master` - /// for snapshot transfer + live streaming. + /// `monoio::net::TcpStream` to + /// `crate::replication::master::handle_psync_inline_single_shard` / + /// `handle_psync_inline_multi_shard` for snapshot transfer + live streaming. HijackForPsync { client_repl_id: String, client_offset: i64, diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index 191cad89..5ab04d05 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/src/server/conn/handler_monoio/txn.rs @@ -408,7 +408,13 @@ pub(super) async fn try_handle_temporal_invalidate( // stream's SELECT context untouched, which stays // truthful: this record neither needs nor changes // the db context. - super::ft::record_local_write(ctx, Bytes::from(record)); + if let Some(rs) = ctx.repl_state.as_ref() { + super::ft::record_local_write( + &rs.read(), + ctx.shard_id, + Bytes::from(record), + ); + } } ctx.shard_databases.wal_append( ctx.shard_id, diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 57e8e9ee..9ba0ec0d 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -272,14 +272,13 @@ pub(super) fn try_handle_replicaof( if let Some(ref rs) = ctx.repl_state { match action { ReplicaofAction::StartReplication { host, port } => { - if let Ok(mut rs_guard) = rs.write() { - rs_guard.set_role(crate::replication::state::ReplicationRole::Replica { + rs.write() + .set_role(crate::replication::state::ReplicationRole::Replica { host: host.clone(), port, state: crate::replication::handshake::ReplicaHandshakeState::PingPending, }); - } let rs_clone = Arc::clone(rs); // Bump the task generation FIRST: any previously spawned // replica task (old REPLICAOF target) sees itself @@ -305,11 +304,10 @@ pub(super) fn try_handle_replicaof( // left it streaming + applying forever (each NO ONE → // re-attach cycle stacked one more live applier). let _ = crate::replication::replica::bump_replica_task_epoch(); - if let Ok(mut rs_guard) = rs.write() { - rs_guard.repl_id2 = rs_guard.repl_id.clone(); - rs_guard.repl_id = generate_repl_id(); - rs_guard.set_role(crate::replication::state::ReplicationRole::Master); - } + let mut rs_guard = rs.write(); + rs_guard.repl_id2 = rs_guard.repl_id.clone(); + rs_guard.repl_id = generate_repl_id(); + rs_guard.set_role(crate::replication::state::ReplicationRole::Master); } ReplicaofAction::NoOp => {} } @@ -349,7 +347,7 @@ pub(super) async fn try_handle_info( } }); if let Some(ref rs) = ctx.repl_state { - if let Ok(rs_guard) = rs.try_read() { + if let Some(rs_guard) = rs.try_read() { response_text.push_str(&crate::replication::handshake::build_info_replication( &rs_guard, )); diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index e871c5d0..0274265a 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -62,7 +62,7 @@ pub(crate) async fn flush_with_aof_ack( mut responses: Vec, aof_entries: Vec<(usize, usize, Bytes)>, pool: &crate::persistence::aof::AofWriterPool, - repl_state: &Option>>, + repl_state: &Option>>, change_counter: &Option>, ) -> bool where @@ -146,7 +146,7 @@ pub async fn handle_connection( runtime_config: Arc>, tracking_table: Arc>, client_id: u64, - repl_state: Option>>, + repl_state: Option>>, acl_table: Arc>, vector_store: Option>>, text_store: Option>>, @@ -868,13 +868,11 @@ pub async fn handle_connection( if let Some(ref rs) = repl_state { match action { ReplicaofAction::StartReplication { host, port } => { - if let Ok(mut rs_guard) = rs.write() { - rs_guard.set_role(crate::replication::state::ReplicationRole::Replica { - host: host.clone(), - port, - state: crate::replication::handshake::ReplicaHandshakeState::PingPending, - }); - } + rs.write().set_role(crate::replication::state::ReplicationRole::Replica { + host: host.clone(), + port, + state: crate::replication::handshake::ReplicaHandshakeState::PingPending, + }); } ReplicaofAction::PromoteToMaster => { use crate::replication::state::generate_repl_id; @@ -882,11 +880,10 @@ pub async fn handle_connection( // bump the generation anyway so any task spawned by // another handler path stops applying. let _ = crate::replication::replica::bump_replica_task_epoch(); - if let Ok(mut rs_guard) = rs.write() { - rs_guard.repl_id2 = rs_guard.repl_id.clone(); - rs_guard.repl_id = generate_repl_id(); - rs_guard.set_role(crate::replication::state::ReplicationRole::Master); - } + let mut rs_guard = rs.write(); + rs_guard.repl_id2 = rs_guard.repl_id.clone(); + rs_guard.repl_id = generate_repl_id(); + rs_guard.set_role(crate::replication::state::ReplicationRole::Master); } ReplicaofAction::NoOp => {} } @@ -952,7 +949,7 @@ pub async fn handle_connection( Frame::BulkString(b) => String::from_utf8_lossy(&b).to_string(), _ => String::new(), }; - if let Ok(rs_guard) = rs.try_read() { + if let Some(rs_guard) = rs.try_read() { response_text.push_str( &crate::replication::handshake::build_info_replication(&rs_guard), ); @@ -965,7 +962,7 @@ pub async fn handle_connection( // --- READONLY enforcement: reject writes on replicas --- if let Some(ref rs) = repl_state { - if let Ok(rs_guard) = rs.try_read() { + if let Some(rs_guard) = rs.try_read() { if matches!( rs_guard.role, crate::replication::state::ReplicationRole::Replica { .. } diff --git a/src/server/conn_state.rs b/src/server/conn_state.rs index d2946f36..10e121b7 100644 --- a/src/server/conn_state.rs +++ b/src/server/conn_state.rs @@ -45,7 +45,7 @@ pub struct ConnectionContext { /// to construct PerShard pools when the on-disk manifest demands it. pub aof_pool: Option>, pub tracking_table: Rc>, - pub repl_state: Option>>, + pub repl_state: Option>>, pub cluster_state: Option>>, pub lua: Rc, pub script_cache: Rc>, diff --git a/src/server/embedded.rs b/src/server/embedded.rs index e011ce5a..d6642b4b 100644 --- a/src/server/embedded.rs +++ b/src/server/embedded.rs @@ -204,7 +204,7 @@ pub async fn run_embedded( let mut rs = crate::replication::state::ReplicationState::new(num_shards, repl_id, repl_id2); rs.set_backlog_capacity(config.repl_backlog_size); - Arc::new(std::sync::RwLock::new(rs)) + Arc::new(RwLock::new(rs)) }; crate::admin::metrics_setup::set_global_repl_state(repl_state.clone()); diff --git a/src/server/listener.rs b/src/server/listener.rs index 81c40569..0cfcbff4 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -192,7 +192,7 @@ pub async fn run_with_shutdown( repl_id, repl_id2, ); rs.set_backlog_capacity(config.repl_backlog_size); - Arc::new(RwLock::new(rs)) + Arc::new(parking_lot::RwLock::new(rs)) }; // Register repl_state globally for INFO command queries. diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index cbd4d327..61a6304e 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -142,7 +142,7 @@ pub(crate) fn spawn_tokio_connection( server_config: &Arc, all_notifiers: &[Arc], snapshot_trigger_tx: &channel::WatchSender, - repl_state: &Option>>, + repl_state: &Option>>, cluster_state: &Option>>, cached_clock: &CachedClock, remote_subscriber_map: &Arc>, @@ -362,7 +362,7 @@ pub(crate) fn spawn_migrated_tokio_connection( server_config: &Arc, all_notifiers: &[Arc], snapshot_trigger_tx: &channel::WatchSender, - repl_state: &Option>>, + repl_state: &Option>>, cluster_state: &Option>>, cached_clock: &CachedClock, remote_subscriber_map: &Arc>, @@ -544,7 +544,7 @@ pub(crate) fn spawn_monoio_connection( server_config: &Arc, all_notifiers: &[Arc], snapshot_trigger_tx: &channel::WatchSender, - repl_state: &Option>>, + repl_state: &Option>>, cluster_state: &Option>>, cached_clock: &CachedClock, remote_subscriber_map: &Arc>, @@ -954,7 +954,7 @@ pub(crate) fn spawn_migrated_monoio_connection( server_config: &Arc, all_notifiers: &[Arc], snapshot_trigger_tx: &channel::WatchSender, - repl_state: &Option>>, + repl_state: &Option>>, cluster_state: &Option>>, cached_clock: &CachedClock, remote_subscriber_map: &Arc>, diff --git a/src/shard/coordinator.rs b/src/shard/coordinator.rs index 3c69c300..6dcb595b 100644 --- a/src/shard/coordinator.rs +++ b/src/shard/coordinator.rs @@ -372,7 +372,7 @@ async fn run_on_owner_persist( /// Type of the replication-state handle threaded into the coordinator's local /// persistence path (same shape `AofWriterPool::issue_append_lsn` expects). type ReplStateRef<'a> = - &'a Option>>; + &'a Option>>; /// Persist a coordinator LOCAL-leg write to the owning shard's AOF, matching the /// local single-key write contract (the `is_write` block in diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 52a6473e..5851f494 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -59,7 +59,7 @@ impl super::Shard { persistence_dir: Option, snapshot_trigger_rx: channel::WatchReceiver, snapshot_trigger_tx: channel::WatchSender, - repl_state_ext: Option>>, + repl_state_ext: Option>>, cluster_state: Option>>, config_port: u16, acl_table: Arc>, @@ -729,25 +729,21 @@ impl super::Shard { // path doesn't need to traverse ReplicationState's outer RwLock per write. let repl_backlog: crate::replication::backlog::SharedBacklog = match repl_state_ext.as_ref() { - Some(rs) => match rs.read() { - Ok(g) => g - .per_shard_backlogs - .get(self.id) - .cloned() - .unwrap_or_else(|| std::sync::Arc::new(parking_lot::Mutex::new(None))), - Err(_) => std::sync::Arc::new(parking_lot::Mutex::new(None)), - }, + Some(rs) => rs + .read() + .per_shard_backlogs + .get(self.id) + .cloned() + .unwrap_or_else(|| std::sync::Arc::new(parking_lot::Mutex::new(None))), None => std::sync::Arc::new(parking_lot::Mutex::new(None)), }; let mut replica_txs: Vec = Vec::new(); - let repl_state: Option>> = repl_state_ext; + let repl_state: Option>> = repl_state_ext; // QW3 (2026-06 review): lock-free offset handle cloned ONCE at shard // startup. The SPSC drain's per-write offset advance goes through this // handle, so the surrounding RwLock is never read-locked per write. - let repl_offsets: Option = repl_state - .as_ref() - .and_then(|rs| rs.read().ok()) - .map(|g| g.offset_handle()); + let repl_offsets: Option = + repl_state.as_ref().map(|rs| rs.read().offset_handle()); // Track last seen snapshot epoch to detect watch channel triggers let mut last_snapshot_epoch = snapshot_trigger_rx.borrow(); diff --git a/src/shard/mq_exec.rs b/src/shard/mq_exec.rs index 3e7d3c94..2924f147 100644 --- a/src/shard/mq_exec.rs +++ b/src/shard/mq_exec.rs @@ -272,7 +272,7 @@ fn replicate_mq_record(shard_id: usize, db_index: usize, cmd_name: &'static [u8] let Some(repl_state) = crate::admin::metrics_setup::get_global_repl_state_arc() else { return; }; - let single_shard = repl_state.read().is_ok_and(|g| g.num_shards() == 1); + let single_shard = repl_state.read().num_shards() == 1; if !single_shard { return; } diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs index 2eb12460..f73c6db8 100644 --- a/src/shard/persistence_tick.rs +++ b/src/shard/persistence_tick.rs @@ -453,7 +453,7 @@ pub(crate) fn run_eviction_tick( pagecache_bytes += mem.pagecache.load(Ordering::Relaxed); } let repl_backlog_bytes = crate::admin::metrics_setup::get_global_repl_state_arc() - .and_then(|state| state.read().ok().map(|g| g.backlog_resident_bytes())) + .map(|state| state.read().backlog_resident_bytes()) .unwrap_or(0); let allocator_overhead = compute_allocator_overhead( rss, diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index af7cef03..411c7f74 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2791,15 +2791,33 @@ pub(crate) fn handle_shard_message_shared( // silently diverge from the handshake-path allocation. crate::replication::state::mark_fanout_active(); let mut guard = repl_backlog.lock(); - if guard.is_none() { - // Seed byte positions at the current shard offset so range - // math stays aligned with pre-attach `issue_lsn` advances — - // see `ReplicationBacklog::new_at`. - let offset = repl_state - .as_ref() - .map(|h| h.shard_offset(shard_id)) - .unwrap_or(0); - *guard = Some(ReplicationBacklog::new_at(backlog_capacity, offset)); + let current_offset = repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0); + match guard.as_mut() { + None => { + // Seed byte positions at the current shard offset so + // range math stays aligned with pre-attach `issue_lsn` + // advances — see `ReplicationBacklog::new_at`. + *guard = Some(ReplicationBacklog::new_at(backlog_capacity, current_offset)); + } + Some(backlog) => { + // Task #70 correctness fix (orchestrator-caught + // regression): this backlog may already have been + // allocated by an earlier bare REPLCONF and skewed + // stale during the FANOUT_HINT-false window — writes in + // that window advanced `current_offset` via the + // bare-`issue_lsn` branch with no matching append. This + // arm runs on the shard's OWN event-loop thread (both + // same-thread self-queue and cross-thread SPSC + // registrations drain here), so this offset read and + // the realign are race-free with this shard's own + // append/advance sequence. Realign BEFORE `cut` / + // `push_offset` are captured below — see + // `ReplicationState::realign_backlog` / `ReplicationBacklog::realign_to`. + backlog.realign_to(current_offset); + } } drop(guard); // Exactly-once cut (per-shard axis): pusher-captured when the @@ -2874,12 +2892,26 @@ pub(crate) fn handle_shard_message_shared( // PSYNC always answers with a full resync). { let mut guard = repl_backlog.lock(); - if guard.is_none() { - let offset = repl_state - .as_ref() - .map(|h| h.shard_offset(shard_id)) - .unwrap_or(0); - *guard = Some(ReplicationBacklog::new_at(backlog_capacity, offset)); + let current_offset = repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0); + match guard.as_mut() { + None => { + *guard = Some(ReplicationBacklog::new_at(backlog_capacity, current_offset)); + } + Some(backlog) => { + // Task #70 correctness fix (orchestrator-caught + // regression): same skew risk and same race-free + // argument as the `RegisterReplica` arm above — this + // whole leg runs synchronously on this shard's own + // thread, BEFORE `shard_offset` (the `cut` for this + // leg) is captured further down. Realigning here + // guarantees the backlog's own `end_offset` matches + // that `cut` exactly, even though this path doesn't + // replay the backlog itself for the snapshot body. + backlog.realign_to(current_offset); + } } } let started = std::time::Instant::now();