perf(replication): per-write fanout gate — parking_lot migration + FANOUT_HINT probe fix + dead PSYNC path deletion (tasks #70, #72)#332
Merged
Conversation
…x; chore(replication): delete dead master PSYNC path (tasks #70, #72) Two pre-soak defects blocking the v0.7.0 tag, fixed together since both touch ReplicationState locking on the per-write hot path. Task #70 — per-write replication fanout gate perf debt (4 sub-fixes): 1. Migrated ReplicationState's outer lock from std::sync::RwLock to parking_lot::RwLock across 27 call sites (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/*). The inner per_shard_backlogs Mutex was already parking_lot; the outer lock was the odd one out, forcing .read().unwrap() / if-let-Ok poisoning dances on the write path. parking_lot doesn't poison: read()/write() return the guard directly, try_read()/try_write() return Option not Result (fixed 3 sites that assumed Result). 2. ensure_backlogs_allocated() — called from try_handle_replconf on ANY bare REPLCONF, including probes and failed handshakes — used to call mark_fanout_active() unconditionally. FANOUT_HINT is a sticky, never-cleared process-global AtomicBool, so one stray REPLCONF permanently taxed every write with the fanout-active gate check even on a server that never completes PSYNC. Split allocation (still load-bearing, kept in ensure_backlogs_allocated) from hint activation (moved to try_handle_psync, on an actual PSYNC/replica registration). 3. record_local_write / record_local_write_db collapsed from up to 3-4 separate repl_state.read() acquisitions per write down to ONE guard, taken at the top of record_local_write_db and threaded through. replication_fanout_active stays a separate short-lived gate lock by design: some external call sites (handler_monoio/mod.rs MOVE/COPY) reach an .await between the gate check and the next repl-state touch, and Rust drops guards at end of lexical scope (not last-use), so folding it into one long-lived guard would hold a lock across .await. Net: at most 2 acquisitions per write (1 gate + 1 write), down from 4-5. 4. Added #[inline] to record_local_write, record_local_write_db, and replication_fanout_active, matching the try_handle_* convention. Task #72 — delete dead master PSYNC path: handle_psync_on_master (both tokio and monoio variants) and register_replica_with_shards in src/replication/master.rs were unreachable — every live PSYNC handler goes through handle_psync_inline_single_shard / handle_psync_inline_multi_shard from shard/conn_accept.rs instead. The dead pair also had a latent broken WAIT implementation: it initialized ack_offsets but never spawned ack_read_loop to drain replica ACKs into them, so wait_for_replicas against that path would block 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. Testing: - Two new unit tests in 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 test_mark_fanout_active_sets_hint. - Lock-across-await audit: every touched call site was checked by reading the actual code, not just grepping. The one risk found (handler_monoio/mod.rs MOVE/COPY awaiting pool.send_append_group() after the fanout gate check) is why replication_fanout_active was kept as its own short-lived guard instead of being folded into the record_local_write_db guard. No parking_lot guard is held across an .await point anywhere in the touched code. - fmt clean; clippy -D warnings clean on both default (monoio) and --no-default-features --features runtime-tokio,jemalloc. - Full cargo test --lib green on both feature sets (4303 passed / 1 ignored monoio; 3484 passed / 1 ignored tokio). - replication_hardening (6/6), replication_multishard (9/9), replication_planes (11/11) integration suites green, MOON_BIN pinned to a fresh release-fast monoio binary, --ignored --test-threads=1. replication_test has 0 tests (empty suite, pre-existing). replication:: unit-test module green under both runtimes (67/67 monoio-filtered, full set included in the --lib totals above) — master-side PSYNC (and thus the suites above) is monoio-only by pre-existing, unmodified design (try_handle_psync_unsupported under tokio); the "both runtimes" gate is satisfied at the unit-test level. - Rebased onto origin/main (108a01e) to pick up PR #329's seed_master_offset shard-0 seeding fix, which touched the same two files (master.rs, state.rs); auto-merged cleanly with no conflict markers, re-verified by re-running every gate above post-rebase. No wire-protocol or observable behavior change beyond the hint gating: REPLCONF -> PSYNC handshake sequence is unchanged. author: Tin Dang
…t regression on task #70) Orchestrator review of eb4d47b3 found a correctness regression in the FANOUT_HINT fix (task #70.2). Splitting backlog allocation from hint activation opened a REPLCONF -> PSYNC window during which a bare REPLCONF could allocate + seed a shard's ReplicationBacklog at its offset at that instant, then -- with FANOUT_HINT still false -- every local write in the window advanced the shard's real offset counter via the bare-issue_lsn branch with NO matching backlog append (the KV wal_append_and_fanout path is unaffected; only the handler-inline issue_append_lsn branch skews). The backlog's end_offset silently drifted 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 (shards=4, appendfsync always, reconnects every 12min). Any snapshot/cut/push-offset captured from that backlog after activation would be range-inconsistent with it: FULLRESYNC catch-up and partial resync could then read wrong bytes, or silently skip catch-up (bytes_from returning None for an offset the backlog didn't yet know about) -- acked-write loss / replica corruption. Fix: restore the invariant "backlog byte positions are trustworthy from the last activation realign onward" by realigning already-allocated backlogs at every activation site, before any snapshot/cut/push offset is captured from them. 1. ReplicationBacklog::realign_to(offset) (src/replication/backlog.rs): resets the buffer to empty, reseeded at offset -- the same state as new_at(capacity, offset). No-op if already aligned. A skewed buffer's contents are already useless (they cover positions the real counter never confirmed), so dropping them 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 serves wrong bytes. 2. ReplicationState::realign_backlog(shard_id) (src/replication/state.rs): reads the shard's current real offset and calls realign_to on that shard's backlog if allocated. Race-free only when called from the shard thread that owns shard_id's own offset advances -- true for every call site below. 3. Wired into try_handle_psync (src/server/conn/handler_monoio/dispatch.rs), right after mark_fanout_active(): realigns ctx.shard_id's backlog. Race-free at shards=1, where the connection task IS the owning shard's own event-loop thread (documented in the code -- this call does NOT cover other shards on a multi-shard master; that's handled by point 4 below, which runs per-shard on each shard's own thread). 4. Wired into the RegisterReplica and PrepareReplicaSync arms (src/shard/spsc_handler.rs): both arms already had an `if guard.is_none() { allocate }` pattern; changed to `match guard.as_mut() { None => allocate, Some(backlog) => realign }` so an already-allocated backlog is realigned instead of silently left skewed, before cut/push_offset/shard_offset are captured further down in the same arm. Testing (RED/GREEN): - backlog.rs: test_realign_to_resets_skewed_backlog, test_realign_to_noop_when_already_aligned -- low-level buffer-reset behavior of realign_to itself. - state.rs: test_realign_backlog_fixes_skew_after_hint_false_window -- reproduces the exact bug: ensure_backlogs_allocated (REPLCONF), then issue_lsn(shard, 100/150/50) with no append (the hint-false window), asserts the skew is real (backlog end_offset stuck at 0 while the real offset is 300, and bytes_from(300) returns None on the unrealigned backlog), then calls realign_backlog and asserts end_offset now matches the real offset and bytes_from(300) returns Some(empty). test_realign_backlog_then_append_reads_back_exact_record is the integration-shaped companion: allocate, issue_lsn, activate, append one record via the normal backlog.append + increment_shard_offset pairing, assert bytes_from(pre-append offset) returns exactly that record's bytes. - RED verified by temporarily neutering realign_backlog's body (#[cfg(any())] stub) and confirming both new state.rs tests fail (end_offset stuck at 0 instead of 300; the post-append record unreadable at its own pre-append offset), then reverted. - GREEN: all 4 new tests pass, plus the full replication:: unit module (71/71) and full cargo test --lib (4307/4307 monoio, 3490/3490 tokio, both 0 failed). Gates: - fmt clean. - clippy -D warnings clean, both default (monoio) and --no-default-features --features runtime-tokio,jemalloc. - cargo test --lib green both feature sets (4307 passed / 1 ignored monoio; 3490 passed / 1 ignored tokio; 0 failed either). - replication_hardening (6/6), replication_multishard (9/9), replication_planes (11/11) integration suites green against a fresh MOON_BIN built from this fix, --ignored --test-threads=1. No VM A/B re-run: the realign call sites are activation-time-only (PSYNC arrival, replica registration), off the per-write hot path the prior A/B measured -- perf surface is unchanged by this fix. author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
📝 WalkthroughWalkthroughReplication state now uses ChangesReplication state lock migration
Fanout gating and backlog alignment
Inline PSYNC lifecycle cleanup
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Replica
participant try_handle_psync
participant ReplicationState
participant ReplicationBacklog
participant handle_psync_inline_multi_shard
participant ShardMessage
Replica->>try_handle_psync: PSYNC
try_handle_psync->>ReplicationState: ensure_backlogs_allocated()
try_handle_psync->>ReplicationState: mark_fanout_active()
try_handle_psync->>ReplicationState: realign_backlog(shard_id)
ReplicationState->>ReplicationBacklog: realign_to(current_offset)
try_handle_psync->>handle_psync_inline_multi_shard: start snapshot and live stream
handle_psync_inline_multi_shard->>ShardMessage: PrepareReplicaSync
ShardMessage->>ReplicationState: register replica
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/replication/reason_del.rs (1)
248-305: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winConsolidate lock acquisitions to reuse a single state guard per write.
The PR objective and AI summary highlight that "replication write recording reuses a single state guard" and "reduces per-write lock acquisitions." However,
push_record_dbcontinues to acquire theparking_lot::RwLockread guard multiple times per write path whenneeds_selectis true (once forneeds_select, and twice more via thepush_recordcalls).Refactoring to read the global state once and pass the reference down significantly reduces lock contention on the hot path.
♻️ Proposed refactor to consolidate lock acquisitions
-fn push_record_db( - repl_state: &Option<std::sync::Arc<parking_lot::RwLock<ReplicationState>>>, - shard_id: usize, - num_shards: usize, - db: usize, - bytes: Bytes, -) { - if num_shards > 1 { - let select = crate::persistence::aof::serialize_select_record(db); - let mut combined = Vec::with_capacity(select.len() + bytes.len()); - combined.extend_from_slice(&select); - combined.extend_from_slice(&bytes); - push_record(repl_state, shard_id, Bytes::from(combined)); - return; - } - let needs_select = repl_state.as_ref().is_some_and(|rs| { - 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 { - push_record( - repl_state, - shard_id, - crate::persistence::aof::serialize_select_record(db), - ); - } - push_record(repl_state, shard_id, bytes); -} - -fn push_record( - repl_state: &Option<std::sync::Arc<parking_lot::RwLock<ReplicationState>>>, - shard_id: usize, - bytes: Bytes, -) { - let mut end_offset = u64::MAX; - if let Some(rs) = repl_state.as_ref() { - 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); - } - crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { - bytes, - end_offset, - }); -} +fn append_to_backlog( + g: &ReplicationState, + shard_id: usize, + bytes: &Bytes, +) -> u64 { + if let Some(slot) = g.per_shard_backlogs.get(shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(bytes); + } + } + g.increment_shard_offset(shard_id, bytes.len() as u64) +} + +fn push_record_db( + repl_state: &Option<std::sync::Arc<parking_lot::RwLock<ReplicationState>>>, + shard_id: usize, + num_shards: usize, + db: usize, + bytes: Bytes, +) { + if num_shards > 1 { + let select = crate::persistence::aof::serialize_select_record(db); + let mut combined = Vec::with_capacity(select.len() + bytes.len()); + combined.extend_from_slice(&select); + combined.extend_from_slice(&bytes); + push_record(repl_state, shard_id, Bytes::from(combined)); + return; + } + + let mut messages = Vec::with_capacity(2); + if let Some(rs) = repl_state.as_ref() { + let g = rs.read(); + let needs_select = 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 + } + }); + + if needs_select { + let select_bytes = Bytes::from(crate::persistence::aof::serialize_select_record(db)); + let end_offset = append_to_backlog(&g, shard_id, &select_bytes); + messages.push((select_bytes, end_offset)); + } + let end_offset = append_to_backlog(&g, shard_id, &bytes); + messages.push((bytes, end_offset)); + } else { + messages.push((bytes, u64::MAX)); + } + + for (b, offset) in messages { + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes: b, + end_offset: offset, + }); + } +} + +fn push_record( + repl_state: &Option<std::sync::Arc<parking_lot::RwLock<ReplicationState>>>, + shard_id: usize, + bytes: Bytes, +) { + let end_offset = if let Some(rs) = repl_state.as_ref() { + append_to_backlog(&rs.read(), shard_id, &bytes) + } else { + u64::MAX + }; + + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes, + end_offset, + }); +}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/replication/reason_del.rs` around lines 248 - 305, Refactor push_record_db and push_record to acquire the ReplicationState read guard once per write and reuse it for stream database selection, backlog appends, and offset advancement. Pass the existing guard or state reference through both push_record calls, including the needs_select path, while preserving the current selection and fanout ordering.src/server/conn/handler_monoio/dispatch.rs (1)
582-644: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winFanout activation + backlog realign happen before PSYNC args are validated.
mark_fanout_active()(line 597) andg.realign_backlog(ctx.shard_id)(line 607) run inside the block at lines 582-608, which executes beforecmd_args[0]/cmd_args[1]are validated as well-formed repl_id/offset (lines 609-644). A PSYNC with exactly 2 args but malformed content (wrongFramevariant, non-UTF8 or non-numeric offset) still triggers the side effects, then fails validation and returns an error.Since
FANOUT_HINTis sticky and never cleared (per the Task#70doc comment two lines above), a single malformed-but-correctly-sized PSYNC frame permanently activates the fanout tax for the process lifetime — exactly the regression this PR's fix was meant to prevent, just via a different trigger (bad PSYNC content instead of bare REPLCONF). It can also discard an already-aligned backlog needed by another connected replica's future partial resync (forcing an unnecessary full resync), thoughrealign_to's described no-op-when-aligned behavior limits this to the skewed case.Move the
repl_id/offsetparsing (currently at 609-644) above the side-effecting block so only a syntactically valid PSYNC can activate fanout/realign the backlog.🐛 Proposed fix: validate before mutating state
let Some(ref rs) = ctx.repl_state else { responses.push(Frame::Error(Bytes::from_static( b"ERR replication is not enabled on this server", ))); return None; }; - { - 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; - } - g.ensure_backlogs_allocated(); - crate::replication::state::mark_fanout_active(); - 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(), _ => { responses.push(Frame::Error(Bytes::from_static( b"ERR PSYNC: invalid replid", ))); return None; } }; let offset_bytes = match &cmd_args[1] { Frame::BulkString(b) | Frame::SimpleString(b) => b.as_ref(), _ => { responses.push(Frame::Error(Bytes::from_static( b"ERR PSYNC: invalid offset", ))); return None; } }; let offset_str = match std::str::from_utf8(offset_bytes) { Ok(s) => s, Err(_) => { responses.push(Frame::Error(Bytes::from_static( b"ERR PSYNC: offset must be an integer", ))); return None; } }; let offset: i64 = match offset_str.parse() { Ok(n) => n, Err(_) => { responses.push(Frame::Error(Bytes::from_static( b"ERR PSYNC: offset must be an integer", ))); return None; } }; + { + 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; + } + g.ensure_backlogs_allocated(); + crate::replication::state::mark_fanout_active(); + g.realign_backlog(ctx.shard_id); + } Some((repl_id, offset)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/server/conn/handler_monoio/dispatch.rs` around lines 582 - 644, Move the repl_id and offset parsing/validation currently following the master-role block to occur before mark_fanout_active and g.realign_backlog in the PSYNC handler. Ensure malformed frame variants, non-UTF8 repl IDs or offsets, and non-numeric offsets return their existing errors without activating fanout or mutating the backlog; preserve the existing side effects for syntactically valid PSYNC commands.
🧹 Nitpick comments (2)
src/replication/state.rs (1)
491-507: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract shared
record_local_writecore logic to avoid drift between the two copies. Both sites implement the identical backlog-append →increment_shard_offset→ReplicaLiveFanout-push sequence, differing only in how theReplicationStateguard is obtained; a future fix to one is easy to miss in the other.
src/replication/state.rs#L491-L507: extract the body (afterlet g = repl_state_arc.read();) into a sharedfn record_local_write_locked(g: &ReplicationState, shard_id: usize, bytes: Bytes)and call it here.src/server/conn/handler_monoio/ft.rs#L92-L110: replace this function's body with a direct call to the newrecord_local_write_lockedhelper fromstate.rs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/replication/state.rs` around lines 491 - 507, Extract the shared backlog-append, shard-offset-increment, and ReplicaLiveFanout dispatch sequence into record_local_write_locked in src/replication/state.rs:491-507, accepting &ReplicationState, shard_id, and Bytes. Update record_local_write_global there to obtain its read guard and call the helper, and replace the duplicate implementation in src/server/conn/handler_monoio/ft.rs:92-110 with a direct call to the shared helper, preserving each function’s existing guard acquisition.src/shard/spsc_handler.rs (1)
383-424: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoffBoth dispatch functions have grown well past the file/function-size guideline. Root cause is the same in both files: a single monolithic
match/if-arm dispatcher that keeps absorbing every new command/message type instead of being split into submodules, pushing both files far beyond the 1500-line file limit (and the function itself past the 1000-line read/write-implementation limit).
src/shard/spsc_handler.rs#L383-L424:handle_shard_message_shared'smatchbody runs to ~line 3046 in a file that is itself ~3047 lines; split per-ShardMessage-variant handling into dedicated submodules (mirrors how command-group files are expected to be split per the guideline).src/server/conn/handler_single.rs#L137-L154:handle_connection's inline dispatch loop runs to ~line 2767; extract the connection-level intercepts (AUTH/HELLO/CLIENT/REPLICAOF/etc.) and the read/write dispatch runs into separate helper modules.Not a blocker for this PR (neither file was restructured here, and the specific changes reviewed above are correct), but worth scheduling before either file grows further.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/shard/spsc_handler.rs` around lines 383 - 424, Refactor the oversized dispatchers into dedicated helper modules: in src/shard/spsc_handler.rs lines 383-424, split handle_shard_message_shared’s ShardMessage-variant handling by message type; in src/server/conn/handler_single.rs lines 137-154, extract handle_connection’s connection-level intercepts and read/write dispatch into separate helpers. Preserve existing behavior while keeping each dispatcher and file within the project size guidelines.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/replication/reason_del.rs`:
- Around line 248-305: Refactor push_record_db and push_record to acquire the
ReplicationState read guard once per write and reuse it for stream database
selection, backlog appends, and offset advancement. Pass the existing guard or
state reference through both push_record calls, including the needs_select path,
while preserving the current selection and fanout ordering.
In `@src/server/conn/handler_monoio/dispatch.rs`:
- Around line 582-644: Move the repl_id and offset parsing/validation currently
following the master-role block to occur before mark_fanout_active and
g.realign_backlog in the PSYNC handler. Ensure malformed frame variants,
non-UTF8 repl IDs or offsets, and non-numeric offsets return their existing
errors without activating fanout or mutating the backlog; preserve the existing
side effects for syntactically valid PSYNC commands.
---
Nitpick comments:
In `@src/replication/state.rs`:
- Around line 491-507: Extract the shared backlog-append,
shard-offset-increment, and ReplicaLiveFanout dispatch sequence into
record_local_write_locked in src/replication/state.rs:491-507, accepting
&ReplicationState, shard_id, and Bytes. Update record_local_write_global there
to obtain its read guard and call the helper, and replace the duplicate
implementation in src/server/conn/handler_monoio/ft.rs:92-110 with a direct call
to the shared helper, preserving each function’s existing guard acquisition.
In `@src/shard/spsc_handler.rs`:
- Around line 383-424: Refactor the oversized dispatchers into dedicated helper
modules: in src/shard/spsc_handler.rs lines 383-424, split
handle_shard_message_shared’s ShardMessage-variant handling by message type; in
src/server/conn/handler_single.rs lines 137-154, extract handle_connection’s
connection-level intercepts and read/write dispatch into separate helpers.
Preserve existing behavior while keeping each dispatcher and file within the
project size guidelines.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 35798bcc-d199-4c62-9de2-c3e57864ad01
📒 Files selected for processing (29)
CHANGELOG.mdsrc/admin/metrics_setup.rssrc/cluster/gossip.rssrc/command/server_admin.rssrc/main.rssrc/persistence/aof/pool.rssrc/replication/backlog.rssrc/replication/master.rssrc/replication/reason_del.rssrc/replication/replica.rssrc/replication/state.rssrc/scripting/bridge.rssrc/server/conn/blocking.rssrc/server/conn/core.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/ft.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_sharded/dispatch.rssrc/server/conn/handler_single.rssrc/server/conn_state.rssrc/server/embedded.rssrc/server/listener.rssrc/shard/conn_accept.rssrc/shard/coordinator.rssrc/shard/event_loop.rssrc/shard/mq_exec.rssrc/shard/persistence_tick.rssrc/shard/spsc_handler.rs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Deep-review Findings 2 & 4 (pre-soak). Two commits:
Commit 1 (tasks #70 + #72):
ReplicationState's outer lock migratedstd::sync::RwLock→parking_lot::RwLock(CLAUDE.md rule) across 27 files; all poisoning.unwrap()/if let Okdances removed; 3try_read/try_writesites fixed to theOptionAPI.FANOUT_HINTno longer flips on a bareREPLCONF(health-check probe / failed handshake permanently taxed every future write with the replication serialize+SPSC round trip). Activation moved to actual PSYNC arrival (try_handle_psync) + per-shard replica registration. Backlog allocation stays on REPLCONF (load-bearing for the handshake window).record_local_write_dbtakes exactly ONErepl_state.read()per call (was up to 3 + the caller's gate read).replication_fanout_activekept as an independent short-lived gate — a MOVE/COPY call site awaitssend_append_groupbetween gate and next touch; folding it in would hold a lock across.await.handle_psync_on_master/register_replica_with_shards/evaluate_psync_shared(~530 lines, zero callers) — the latent broken-WAIT hazard path.Commit 2 (orchestrator-review-caught regression fix): splitting allocation from hint activation opened a REPLCONF→PSYNC window where handler-inline writes advanced shard offsets via bare
issue_append_lsnwith no matching backlog append → permanent byte/offset skew → FULLRESYNC catch-up / partial resync could serve wrong bytes or silently skip (acked-write loss on replica attach under load — the exact 24h-soak shape). Fix:ReplicationBacklog::realign_to+ReplicationState::realign_backlog, wired intotry_handle_psyncand both spsc registration arms (owning-shard-thread, before cut/push_offset capture). Skewed buffers reset-to-empty at the real offset; stale partial-resync offsets fail safe to full resync.Verification
test_ensure_backlogs_allocated_does_not_activate_fanout_hint(RED via bug reintroduction),test_realign_backlog_fixes_skew_after_hint_false_window+test_realign_backlog_then_append_reads_back_exact_record(RED via neutered realign), + 3 more.cargo test --lib: 4307/4307 monoio, 3490/3490 tokio. fmt + both clippy matrices clean.--ignored --test-threads=1):replication_hardening6/6,replication_multishard9/9,replication_planes11/11.Pre-soak: this fix must be in the soaked binary (REPL-SOAK-01).
Summary by CodeRabbit