feat(replication): R2 multi-shard master PSYNC — merged RDB + per-record SELECT framing#283
Merged
Merged
Conversation
…ord SELECT framing Masters running --shards N now serve full replication to a single-shard replica (task #20, RFC 1B). Previously PSYNC on a multi-shard master was rejected outright. Design: - ShardMessage::PrepareReplicaSync fans to every shard (self queue for the accepting shard, SPSC mesh + notifier for the rest). Each shard's arm serializes its keyspace slice to an RDB BODY, captures its shard offset, and registers the replica's live channel in ONE synchronous stretch on its own thread — per shard there is no window between "in the snapshot" and "streamed live", so no backlog catch-up leg exists and non-idempotent commands cannot double-apply. - The PSYNC task stitches bodies into ONE valid Redis-format RDB (redis_rdb::write_rdb_merged) and replies `+FULLRESYNC <replid> <sum of shard offsets>` + one $<len> bulk — the replica's existing R0 loader is unchanged. Index defs ride once; graph content is sharded so the snapshot carries one moon-graph-store aux entry PER shard, and the replica imports all of them (install_graph_store_many / read_moon_aux_all). - Merged-wire db context: N shard threads feed one replica socket, so per-shard SELECT tracking cannot work. On multi-shard masters every db-scoped record is fused with its own `SELECT <db>` prefix as ONE record (one channel send, one backlog append pair, one offset advance) in both the cross-shard path (wal_append_and_fanout) and the local leg (record_local_write_db) — no interleave can split a SELECT from the write it frames. Gated on the replica-attach hint; single-shard masters keep the emit-on-change tracking. - Partial resync degrades to full: a single scalar offset cannot map back onto N per-shard backlogs. - R1 pieces carry over unchanged: overflow-kick (shared kicked flag across all N fan-out entries), REPLCONF ACK reader, WAIT (summed snapshot offset keeps total_offset - base == bytes-on-wire, so ACK math stays exact). Verification (red/green): - NEW tests/replication_multishard.rs — 6 e2e over real processes, written first and failing on main: 2/4/8-shard full resync + live convergence with INCR exactness, interleaved multi-db writers (5k keys x 2 dbs, leak asserts both directions), per-shard graph snapshot import, raw-handshake partial->full degradation with merged-RDB REDIS-magic assert. - NEW unit merged_multishard_rdb_round_trip (repeated SELECTDB sections + repeated graph aux entries load as one keyspace, CRC valid). - Regression: replication_streaming 7/7, replication_hardening 3/3, replication_graph 5/5, aof_multidb_kill9 4/4, lib 4115 (monoio) + 3316 (tokio), clippy -D warnings on both matrices, fmt clean. author: Tin Dang <tindang.ht97@gmail.com>
…areReplicaSync arm author: Tin Dang <tindang.ht97@gmail.com>
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? |
|
Warning Review limit reached
Next review available in: 49 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughMulti-shard masters now support full PSYNC to single-shard replicas through merged per-shard snapshots, repeated AUX restoration, offset-based live fan-out, and runtime-specific PSYNC handling. Replica task epochs prevent obsolete replication tasks from applying data after reattachment or promotion. ChangesMulti-shard replication
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Replica
participant PSYNCHandler
participant ShardWorkers
participant RDBMerger
participant SelfQueue
Replica->>PSYNCHandler: PSYNC request
PSYNCHandler->>ShardWorkers: PrepareReplicaSync
ShardWorkers-->>PSYNCHandler: PreparedShardSync bodies and offsets
PSYNCHandler->>RDBMerger: write_rdb_merged
RDBMerger-->>Replica: +FULLRESYNC and merged RDB
SelfQueue-->>Replica: cut-filtered live records
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 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 |
RFC v0.2-R3 (2A + 4B), riding the R2 PR:
- A runtime-tokio master now answers PSYNC with
`-ERR PSYNC requires runtime-monoio on the master (this build runs
runtime-tokio)` instead of falling through to generic dispatch — an
attaching replica's log states WHY the sync failed. Wired in both tokio
paths (handler_sharded intercept + handler_single inline arm) and verified
live against real tokio binaries at shards=1 and shards=2.
- Docs refreshed for the R2 topology:
- docs/guides/clustering.md: v0.1.x "master must run --shards 1" warning
replaced with the v0.7 deployment shape (any-N master / --shards 1
replicas, merged-RDB semantics, partial->full at N>1, WS/MQ plane
caveat).
- README.md: replication bullets updated (also fixes the stale claim that
replicas could run --shards N — the replica task refuses N != 1).
- docs/PRODUCTION-CONTRACT.md: REPL-MULTISHARD-01 and WAIT-01 flipped to
done with R2/R1 evidence.
- CHANGELOG [Unreleased] amended accordingly.
author: Tin Dang <tindang.ht97@gmail.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
tests/replication_multishard.rs (1)
476-476: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUnnecessary single-element tuple destructuring.
let (master_port,) = (17061,);uses a single-element tuple where a simple binding suffices.✨ Suggested cleanup
- let (master_port,) = (17061,); + let master_port = 17061;🤖 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 `@tests/replication_multishard.rs` at line 476, Replace the single-element tuple destructuring in the master_port initialization with a direct binding to 17061, preserving the existing value and subsequent uses of master_port.src/persistence/redis_rdb.rs (1)
578-610: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
read_moon_aux_alllacks a dedicated truncated-buffer test.The existing
read_moon_aux_rejects_foreign_and_truncated_bufferstest (line 896) coversread_moon_auxbut notread_moon_aux_all. Sinceread_moon_aux_allhas the same truncation/foreign-buffer surface, a symmetric test would guard against regressions if the two functions diverge in the future.🧪 Suggested test addition
#[test] fn read_moon_aux_all_rejects_foreign_and_truncated_buffers() { assert!(read_moon_aux_all(b"", MOON_AUX_GRAPH_STORE).is_empty()); assert!(read_moon_aux_all(b"REDIS", MOON_AUX_GRAPH_STORE).is_empty()); assert!(read_moon_aux_all(b"NOTRDB0010....", MOON_AUX_GRAPH_STORE).is_empty()); // Plain RDB with no moon aux: header walk ends at first non-AUX opcode. let db = Database::new(); let mut buf = Vec::new(); write_rdb_refs(&[&db], &mut buf); assert!(read_moon_aux_all(&buf, MOON_AUX_GRAPH_STORE).is_empty()); // Truncated mid-aux must not panic and must return partial results safely. let db2 = Database::new(); let mut buf2 = Vec::new(); write_rdb_refs_with_moon_aux( &[&db2], &[(MOON_AUX_GRAPH_STORE, b"blob")], &mut buf2, ); for cut in 9..buf2.len().min(40) { let _ = read_moon_aux_all(&buf2[..cut], MOON_AUX_GRAPH_STORE); } }🤖 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/persistence/redis_rdb.rs` around lines 578 - 610, Add a dedicated test for read_moon_aux_all covering both foreign and truncated RDB buffers, mirroring read_moon_aux_rejects_foreign_and_truncated_buffers. Assert that each invalid input returns an empty Vec, while leaving the implementation unchanged.
🤖 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.
Inline comments:
In `@src/replication/master.rs`:
- Around line 828-836: Update handle_psync_inline_multi_shard to return the
replication module’s thiserror-based error type instead of anyhow::Result, and
replace any anyhow-specific error construction or propagation within this path
with the typed replication error. Preserve conversion to anyhow only at the
application boundary, not inside library code.
- Around line 854-913: Update the PrepareReplicaSync preparation flow around
reply collection and its later socket-write exits to enforce a timeout for every
recv_async() wait, and add an all-exit cleanup guard covering both self and
remote shards. Ensure the guard unregisters any shards that successfully
prepared whenever preparation fails, times out, or socket writing returns early,
while preserving normal cleanup after a successful sync.
- Around line 847-852: The merged live channel created near replica_id with
mpsc_bounded::<bytes::Bytes> must not serve as the only buffer while the RDB
snapshot is being prepared and written. Add a bounded, configurable pre-RDB
staging strategy and update the registration/snapshot flow around the rx drain
loop (including the logic near the additionally referenced section) so writes
accumulated before draining do not immediately trigger replica eviction, while
preserving bounded memory and existing post-RDB live-channel behavior.
In `@src/shard/spsc_handler.rs`:
- Around line 2587-2683: Extract the replica registration, preparation, and
fan-out handling from the oversized shard handler into a dedicated replication
module. Move the PrepareReplicaSync arm and its related helper logic while
preserving behavior, state updates, reply handling, and fan-out registration;
leave the main dispatch flow delegating to the new module and ensure the
original file is reduced below the 1500-line limit.
---
Nitpick comments:
In `@src/persistence/redis_rdb.rs`:
- Around line 578-610: Add a dedicated test for read_moon_aux_all covering both
foreign and truncated RDB buffers, mirroring
read_moon_aux_rejects_foreign_and_truncated_buffers. Assert that each invalid
input returns an empty Vec, while leaving the implementation unchanged.
In `@tests/replication_multishard.rs`:
- Line 476: Replace the single-element tuple destructuring in the master_port
initialization with a direct binding to 17061, preserving the existing value and
subsequent uses of master_port.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 61490172-4654-41b7-84c3-eed8f4675b91
📒 Files selected for processing (12)
CHANGELOG.mdsrc/persistence/redis_rdb.rssrc/replication/apply.rssrc/replication/graph_sync.rssrc/replication/master.rssrc/replication/state.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/ft.rssrc/shard/conn_accept.rssrc/shard/dispatch.rssrc/shard/spsc_handler.rstests/replication_multishard.rs
Comment on lines
+828
to
+836
| pub async fn handle_psync_inline_multi_shard( | ||
| mut stream: monoio::net::TcpStream, | ||
| repl_state: Arc<RwLock<ReplicationState>>, | ||
| replica_addr: std::net::SocketAddr, | ||
| dispatch_tx: Rc<RefCell<Vec<ringbuf::HeapProd<crate::shard::dispatch::ShardMessage>>>>, | ||
| spsc_notifiers: Vec<std::sync::Arc<crate::runtime::channel::Notify>>, | ||
| self_shard_id: usize, | ||
| num_shards: usize, | ||
| ) -> anyhow::Result<()> { |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Use the replication module’s typed error instead of anyhow.
The new library API introduces anyhow::Result and anyhow construction throughout this path. Return a thiserror-based replication error and convert it at the application boundary.
As per coding guidelines, “Use anyhow only in main.rs and test code; library code must use thiserror or Frame::Error.”
🤖 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/master.rs` around lines 828 - 836, Update
handle_psync_inline_multi_shard to return the replication module’s
thiserror-based error type instead of anyhow::Result, and replace any
anyhow-specific error construction or propagation within this path with the
typed replication error. Preserve conversion to anyhow only at the application
boundary, not inside library code.
Source: Coding guidelines
Comment on lines
+2587
to
+2683
| ShardMessage::PrepareReplicaSync(payload) => { | ||
| // R2 (task #20): this shard's leg of a multi-shard full resync. | ||
| // The ENTIRE leg — RDB body serialization, offset capture, live | ||
| // fan-out registration — runs in this one synchronous stretch on | ||
| // the shard's own thread, so no mutation can slip between "inside | ||
| // the snapshot" and "delivered live" (the same atomicity argument | ||
| // as `handle_psync_inline_single_shard`, applied per shard). | ||
| // | ||
| // This additionally leans on the self-queue-FIRST drain order | ||
| // (see `drain_spsc_shared`): a local write visible to this body | ||
| // capture pushed its `ReplicaLiveFanout` BEFORE this arm could | ||
| // drain, and the self queue drains first — so that fan-out | ||
| // message no-ops against the not-yet-registered replica instead | ||
| // of double-delivering a record that is already in the body. | ||
| let crate::shard::dispatch::PrepareReplicaSyncPayload { | ||
| replica_id, | ||
| tx, | ||
| kicked, | ||
| backlog_capacity, | ||
| reply_tx, | ||
| } = *payload; | ||
| crate::replication::state::mark_fanout_active(); | ||
| // Lazy backlog init (offset accounting parity with RegisterReplica; | ||
| // the backlog itself is not replayed on this path — multi-shard | ||
| // 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 started = std::time::Instant::now(); | ||
| let mut rdb_body: Vec<u8> = Vec::new(); | ||
| let mut vector_defs: Option<Vec<u8>> = None; | ||
| let mut text_defs: Option<Vec<u8>> = None; | ||
| #[cfg(feature = "graph")] | ||
| let mut graph_blob: Vec<u8> = Vec::new(); | ||
| crate::shard::slice::with_shard(|s| { | ||
| let refs: Vec<&crate::storage::Database> = s.databases.iter().collect(); | ||
| crate::persistence::redis_rdb::write_rdb_body_refs(&refs, &mut rdb_body); | ||
| // Index DEFINITIONS ride as moon aux (same as the single-shard | ||
| // path); contents stay in sync via the live stream + backfill. | ||
| let pairs = s.vector_store.collect_index_metas_with_weights(); | ||
| if !pairs.is_empty() { | ||
| vector_defs = Some(crate::vector::index_persist::serialize_index_metas_v5( | ||
| &pairs, | ||
| )); | ||
| } | ||
| let metas = s.text_store.collect_index_metas(); | ||
| if !metas.is_empty() { | ||
| text_defs = Some(crate::text::index_persist::serialize_text_index_metas( | ||
| &metas, | ||
| )); | ||
| } | ||
| #[cfg(feature = "graph")] | ||
| { | ||
| graph_blob = | ||
| crate::replication::graph_sync::export_graph_store(&mut s.graph_store); | ||
| } | ||
| }); | ||
| let shard_offset = repl_state | ||
| .as_ref() | ||
| .map(|h| h.shard_offset(shard_id)) | ||
| .unwrap_or(0); | ||
| replica_txs.push((replica_id, tx, kicked)); | ||
| tracing::debug!( | ||
| shard_id, | ||
| replica_id, | ||
| body_bytes = rdb_body.len(), | ||
| shard_offset, | ||
| elapsed_ms = started.elapsed().as_millis() as u64, | ||
| "prepared multi-shard replica sync leg" | ||
| ); | ||
| let prepared = crate::shard::dispatch::PreparedShardSync { | ||
| rdb_body, | ||
| shard_offset, | ||
| vector_defs, | ||
| text_defs, | ||
| #[cfg(feature = "graph")] | ||
| graph_blob, | ||
| }; | ||
| if reply_tx.try_send(prepared).is_err() { | ||
| // The PSYNC task is gone (replica dropped mid-handshake) — | ||
| // undo the registration so this shard doesn't fan out to a | ||
| // channel nobody drains. | ||
| replica_txs.retain(|(id, _, _)| *id != replica_id); | ||
| tracing::warn!( | ||
| shard_id, | ||
| replica_id, | ||
| "PrepareReplicaSync reply dropped — replica disconnected before sync" | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Extract the replication arms from this oversized module.
This file exceeds 3,600 lines; adding another substantial protocol arm compounds the existing repository-limit violation. Move replica registration/preparation and fan-out helpers into a dedicated module.
As per coding guidelines, “No single Rust file should exceed 1500 lines.”
🤖 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 2587 - 2683, Extract the replica
registration, preparation, and fan-out handling from the oversized shard handler
into a dedicated replication module. Move the PrepareReplicaSync arm and its
related helper logic while preserving behavior, state updates, reply handling,
and fan-out registration; leave the main dispatch flow delegating to the new
module and ensure the original file is reduced below the 1500-line limit.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
docs/guides/clustering.md (1)
36-48: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winStale "leader must be --shards 1" section contradicts the new deployment-shape box above it.
The freshly-updated info box (lines 17-22) states masters support any
--shards N, but this immediately-following section still says the leader "must be --shards 1 in v0.1.x" and only shows a single-shard leader example. This will confuse a reader following the guide top-to-bottom.📝 Proposed fix
-### Start the leader (must be --shards 1 in v0.1.x) +### Start the leader (any --shards N)🤖 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 `@docs/guides/clustering.md` around lines 36 - 48, Update the “Start the leader” subsection in the clustering guide to remove the stale “must be --shards 1 in v0.1.x” restriction and show a leader command consistent with masters supporting any shard count. Keep the replica setup example and surrounding deployment instructions unchanged.
🧹 Nitpick comments (1)
src/server/conn/handler_sharded/mod.rs (1)
864-868: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftBoth connection-loop files far exceed the 1500-line coding-guideline ceiling.
handler_sharded/mod.rsis over 2260 lines andhandler_single.rsis over 2700 lines; the guideline directs splitting command-group files into directory modules as they approach the limit (as already done forhandler_sharded/dispatch.rs's helper functions). This PR's PSYNC branches add a few more lines to each, marginally worsening an existing violation.
src/server/conn/handler_sharded/mod.rs#L864-L868: extract more of the connection-level command-intercept chain (already partially split intodispatch.rs) into further per-concern modules to bringhandle_connection_sharded_innerunder the limit.src/server/conn/handler_single.rs#L862-L869: apply the same command-dispatch-module extraction pattern used inhandler_sharded/dispatch.rsto shrink this file.As per coding guidelines: "No single Rust file should exceed 1500 lines. Split command-group files into directory modules when approaching the limit; split read and write implementations above 1000 lines."
🤖 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_sharded/mod.rs` around lines 864 - 868, The connection handlers exceed the 1500-line guideline and need command-dispatch extraction. In src/server/conn/handler_sharded/mod.rs lines 864-868, move additional command-intercept logic from handle_connection_sharded_inner into per-concern modules following handler_sharded/dispatch.rs; in src/server/conn/handler_single.rs lines 862-869, apply the same dispatch-module pattern. Preserve existing PSYNC and other command-handling behavior while bringing both files below the guideline limit.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.
Inline comments:
In `@docs/guides/clustering.md`:
- Around line 14-34: The clustering guide’s runtime-tokio PSYNC error text is
incomplete. Update the quoted error in the “Supported deployment shape” section
to include the full parenthetical “(this build runs runtime-tokio)” while
preserving the existing `-ERR` protocol prefix and surrounding guidance.
---
Outside diff comments:
In `@docs/guides/clustering.md`:
- Around line 36-48: Update the “Start the leader” subsection in the clustering
guide to remove the stale “must be --shards 1 in v0.1.x” restriction and show a
leader command consistent with masters supporting any shard count. Keep the
replica setup example and surrounding deployment instructions unchanged.
---
Nitpick comments:
In `@src/server/conn/handler_sharded/mod.rs`:
- Around line 864-868: The connection handlers exceed the 1500-line guideline
and need command-dispatch extraction. In src/server/conn/handler_sharded/mod.rs
lines 864-868, move additional command-intercept logic from
handle_connection_sharded_inner into per-concern modules following
handler_sharded/dispatch.rs; in src/server/conn/handler_single.rs lines 862-869,
apply the same dispatch-module pattern. Preserve existing PSYNC and other
command-handling behavior while bringing both files below the guideline limit.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b69abc1e-d116-44dc-98dc-fc27e8c888e0
📒 Files selected for processing (7)
CHANGELOG.mdREADME.mddocs/PRODUCTION-CONTRACT.mddocs/guides/clustering.mdsrc/server/conn/handler_sharded/dispatch.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_single.rs
✅ Files skipped from review due to trivial changes (2)
- README.md
- CHANGELOG.md
…k epoch cancellation Attach-under-write stress testing of R2 (#283) surfaced three defects. All are fixed here before the PR merges; none shipped in a release. 1) REPLICAOF leaked the previous replica task (pre-existing, dominant signal). `REPLICAOF host port` spawned a fresh run_replica_task without stopping the old one, and `REPLICAOF NO ONE` only flipped the role: the old task kept streaming AND applying. Each NO-ONE → re-attach cycle stacked one more live applier — replica INCR counters ran ~25-35% ABOVE the master (reproduced at shards=1 and shards=4). Fix: process-global REPLICA_TASK_EPOCH ticket. StartReplication and PromoteToMaster bump the generation; a superseded task exits at the reconnect-loop top, before snapshot load, and before applying any parsed chunk — it can never mutate the keyspace again. 2) Snapshot-vs-live exactly-once now rests on an offset cut, not on FIFO placement. Two review rounds found opposite failure modes for placement schemes on the PSYNC connection's own shard: - queued snapshot leg (drain-time capture): a local write between push and drain was in the body AND live-sent behind the registration → double-apply (reproduced, +35%); - inline capture + queued registration: a same-cycle SPSC execute was neither in the body (applied after capture) nor live-sent (execute_batch direct-send runs before other_messages processes the registration) — with its bytes in the backlog and the offset advanced: silent loss + permanent replica ACK lag. Fix: every fan-out entry records cut = shard offset at body capture; every live record carries its per-shard end_offset; delivery requires end_offset > cut. Registration placement no longer matters, so ALL shards — including the PSYNC connection's own — use the same PrepareReplicaSync arm (the special-cased inline self-leg is gone). The cut/end_offset axis is the PER-SHARD counter, never the master offset (seed_master_offset diverges the two after AOF recovery). 3) Same-key wire ordering. SPSC-dispatched writes sent to replicas directly from the execute arm while local handler writes deferred through the self queue — a later-offset write could reach the wire before an earlier-offset same-key write, replaying out of the master's serialization order (analysis finding; not reproduced in ~10 black-box attempts). ALL live sends now flow through the self-queue ReplicaLiveFanout arm: per-shard wire order == FIFO order == offset order by construction. Tests (red/green: 1 and 2 reproduced red before their fixes): - tests/replication_multishard.rs +3: attach_under_write_no_double_apply (4-shard, 5x detach/re-attach under 4-writer pipelined INCR load, exact per-counter parity), singleshard_master_attach_under_write_control (same at shards=1 — the discriminating control that proved the task leak pre-existing), same_key_write_order_parity (12 conns APPEND-race 32 shared keys through both write paths; replica must byte-equal master). - Full regression: multishard 9/9, streaming 7/7, hardening 5/5, graph 5/5, aof_multidb_kill9 4/4, lib 4115 (monoio) + 3316 (tokio), clippy -D warnings both matrices, fmt — macOS and Linux VM. Mechanical: ShardMessage::RegisterReplica boxed (RegisterReplicaPayload; the extra offset fields pushed it past the 64-byte cap), ReplicaFanout tuple → struct with cut, increment_shard_offset returns the per-shard post-advance offset. Test-harness hardening surfaced by the Linux VM sweep (task #18 class): - replication_hardening now honors MOON_BIN (hardcoded ./target/release/moon exec'd the wrong-platform binary on the shared macOS/Linux checkout — all 5 tests failed to connect on the VM); - aof_multidb_kill9 wait_ready treats a connection RESET during moon's bootstrap→per-shard SO_REUSEPORT listener handover as "not ready yet" instead of panicking (3/4 VM failures, reproduced with main's binary too — environment-timing flake, not a code regression). Refs: task #20, task #36, PR #283 author: Tin Dang <tindang.ht97@gmail.com>
… failure (review) CodeRabbit round on PR #283: - PrepareReplicaSync reply collection is now bounded (30s timeout per leg): a wedged shard can no longer park the PSYNC task — and the fan-out registrations it holds — forever. On expiry the sync aborts with explicit unregister everywhere; the replica reconnects and retries. - Socket-write failures during the +FULLRESYNC/RDB transfer also unregister on every shard instead of leaving the entries to the passive next-write Disconnected prune. - docs/guides/clustering.md quotes the full tokio PSYNC error text. - CHANGELOG documents the 16K pre-RDB live-buffer limitation (overflow during a very large snapshot under sustained writes KICKS the replica loudly into a retry — never silent divergence). Verified: multishard 9/9, clippy -D warnings, fmt. Refs: task #20, PR #283 author: Tin Dang <tindang.ht97@gmail.com>
Contributor
Author
|
Review round addressed in 6839aec:
Also since the last review: commit 8e56d23 fixes two attach-under-write P0s found by stress testing (REPLICAOF replica-task leak → epoch cancellation; snapshot-vs-live exactly-once → per-replica offset cut) — see the PR body Round 2 section. |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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.
Inline comments:
In `@src/replication/replica.rs`:
- Around line 265-267: Move the superseded(cfg.epoch) check before the
FULLRESYNC state write in both src/replication/replica.rs:265-267 and
src/replication/replica.rs:596-598. Ensure each path checks and bails before
updating repl_id, master_repl_offset, ReplicaHandshakeState::FullResyncLoading,
or persisted replication state; no direct change is needed beyond relocating the
existing guard in each site.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 51610959-d2ea-4d0f-a5bf-3274269a64f2
📒 Files selected for processing (13)
CHANGELOG.mdsrc/replication/master.rssrc/replication/replica.rssrc/replication/state.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/ft.rssrc/server/conn/handler_sharded/dispatch.rssrc/server/conn/handler_single.rssrc/shard/dispatch.rssrc/shard/spsc_handler.rstests/aof_multidb_kill9.rstests/replication_hardening.rstests/replication_multishard.rs
✅ Files skipped from review due to trivial changes (1)
- CHANGELOG.md
Comment on lines
+265
to
+267
| if superseded(cfg.epoch) { | ||
| anyhow::bail!("replica task superseded before snapshot load"); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the file structure first, then inspect the relevant regions.
ast-grep outline src/replication/replica.rs --view expanded || true
printf '\n--- tokio region ---\n'
sed -n '210,290p' src/replication/replica.rs
printf '\n--- monoio region ---\n'
sed -n '540,620p' src/replication/replica.rs
printf '\n--- superseded uses ---\n'
rg -n "superseded\\(|repl_state|master_repl_offset|repl_id|Replica" src/replication/replica.rsRepository: pilotspace/moon
Length of output: 14641
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the task-epoch and handshake flow around the reconnect loop.
sed -n '1,180p' src/replication/replica.rs
printf '\n--- tokio full handshake section ---\n'
sed -n '220,310p' src/replication/replica.rs
printf '\n--- monoio full handshake section ---\n'
sed -n '565,640p' src/replication/replica.rs
printf '\n--- epoch bump / superseded definitions ---\n'
rg -n "bump_replica_task_epoch|REPLICA_TASK_EPOCH|fn superseded|superseded\\(" src/replication/replica.rsRepository: pilotspace/moon
Length of output: 15497
Move the superseded check before the FULLRESYNC state write in both paths.
src/replication/replica.rs:265src/replication/replica.rs:596
A superseded task can still overwrite repl_id, master_repl_offset, ReplicaHandshakeState::FullResyncLoading, and the persisted replication state before it bails.
📍 Affects 1 file
src/replication/replica.rs#L265-L267(this comment)src/replication/replica.rs#L596-L598
🤖 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/replica.rs` around lines 265 - 267, Move the
superseded(cfg.epoch) check before the FULLRESYNC state write in both
src/replication/replica.rs:265-267 and src/replication/replica.rs:596-598.
Ensure each path checks and bails before updating repl_id, master_repl_offset,
ReplicaHandshakeState::FullResyncLoading, or persisted replication state; no
direct change is needed beyond relocating the existing guard in each site.
pilotspacex-byte
added a commit
that referenced
this pull request
Jul 11, 2026
…rver-spawning suites (#284) * test(harness): shared spawn_listening helper + spsc_two_db reference conversion (task #18 WIP) * test(harness): sweep 33 suites onto shared TOCTOU-safe port/spawn helpers Task #18. Every integration suite spawning a real moon process carried its own free_port() that binds :0 and DROPS the listener before the server spawns. Two failure modes kept hitting CI (latest: spsc_two_db on PR #283, which cost two full macOS rerolls): 1. Port TOCTOU — between probe drop and moon's bind, another test's probe or a concurrent outbound connection's ephemeral source port takes the port; moon exits with EADDRINUSE. 2. Dead-server blind poll — harnesses polled connect() for up to 30s without checking child liveness, so a lost bind race surfaced as "server never accepted: Connection refused" half a minute later with the real error unread in the server's stderr log. Fix: new shared tests/common/mod.rs with - reserve_port(): process-wide dedup set over kernel-chosen probe ports (kills intra-process reuse); - spawn_listening(spawn): polls TCP accept WHILE watching child.try_wait(), respawning on a fresh port the moment the child dies (external steals cannot be prevented, only recovered from; 3 attempts, then a loud panic pointing at the server's stderr log). All 33 suites converted. Conversion rules held throughout: - protocol-level readiness (PING/AUTH) stays with each suite — spawn_listening only guarantees "listening"; - kill-9/SIGTERM/restart tests keep their deliberate same-port+same-dir restart legs (only the FIRST spawn of a lifecycle goes through spawn_listening): coordinator_local_leg_durability (7 legs), sharded_multi_exec_durability, sharded_multi_exec_routing, spsc_wake_floor_red, vector_db_isolation; - expected-startup-failure tests (db_maxmemory_quota CLI validation, admin_auth hard02) keep direct spawns on reserve_port() ports; - txn_kv_wiring's in-process async server routes port choice through reserve_port() and keeps its await_server_ready poll (no Child to watch); its real-subprocess crash-recovery path uses spawn_listening; - no test assertions, timeouts, or CLI flags changed. Found en route (filed separately, not fixed here): admin_auth_cors_ratelimit is cfg(feature = "console")-gated and has 13 pre-existing compile errors from ureq API drift — invisible because the console CI job is skipped on PRs (task #37). Verified: cargo test --no-run clean on default AND tokio matrices; all 33 suites green (--no-fail-fast); 10-rep stress running 14 port-hungry suites CONCURRENTLY (the CI contention pattern) green; fmt clean. Refs: task #18, follow-up to PR #283 CI rerolls author: Tin Dang <tindang.ht97@gmail.com> --------- Co-authored-by: Tin Dang <tindang.ht97@gmail.com>
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
R2 (task #20, RFC 1B): a master running
--shards Nnow serves full replication to a single-shard replica. Previously any PSYNC against a multi-shard master was rejected with-ERR PSYNC across multiple shards is not yet supported.Design
ShardMessage::PrepareReplicaSyncfans to every shard — the accepting shard via the self queue, the rest over the SPSC mesh + notifier. Each shard's arm serializes its keyspace slice to an RDB body, captures its shard replication offset, and registers the replica's live channel in one synchronous stretch on its own thread: per shard there is no window between "inside the snapshot" and "streamed live", so no backlog catch-up leg exists and non-idempotent commands (INCR) cannot double-apply. This also leans on the existing self-queue-first drain invariant (documented in the arm).redis_rdb::write_rdb_merged: header + moon aux + bodies + EOF/CRC64) and answers+FULLRESYNC <replid> <Σ shard offsets>with one$<len>bulk. The replica's existing R0 loader is unchanged. Index definitions ride once; graph content is sharded, so the snapshot carries onemoon-graph-storeaux entry per shard and the replica imports all of them (install_graph_store_many,read_moon_aux_all).SELECT <db>prefix as ONE record (one channel send / one backlog append pair / one offset advance) in both the cross-shard path (wal_append_and_fanout) and the local leg (record_local_write_db) — no cross-shard interleave can split a SELECT from the write it frames. Gated on the replica-attach hint; single-shard masters keep the cheaper emit-on-change tracking.+FULLRESYNC.Arcshared across all N shards' fan-out entries;REPLCONF ACK+WAITstay exact because the summed snapshot offset keepstotal_offset − base == bytes on wire.Verification (red/green + standing replication gates)
New tests (written first, failing on main):
tests/replication_multishard.rs— 6 e2e over real processes: 2/4/8-shard full resync + live convergence with INCR exactness; interleaved multi-db writers (5k keys × 2 dbs, pipelined, leak asserts both directions); per-shard graph snapshot import; raw-handshake partial→full degradation with merged-RDBREDISmagic assert. Green on macOS and the Linux VM (monoio intercept paths are CI-blind, so both platforms were run explicitly).merged_multishard_rdb_round_trip— repeated SELECTDB sections + repeated graph aux entries load as one keyspace, CRC valid.Regression: replication_streaming 7/7 · replication_hardening 3/3 · replication_graph 5/5 · aof_multidb_kill9 4/4 (kill-9 durability gate) · lib 4115 (monoio) + 3316 (tokio) · clippy
-D warningsboth matrices · fmt.VM A/B bench (OrbStack Linux, 2 reps, main 4ef0b21 vs branch):
Every replica-attached run converged to exact DBSIZE parity (e.g. 98233/98233) after the 200k random-key benchmark, on both binaries at shards=1 and on the branch at shards=4.
Known limitations (documented in CHANGELOG)
runtime-monoio(the default); the tokio build has no master-side PSYNC intercept (pre-existing).--shards 1replicas).Closes task #20 (v0.7 R2).
Round 2 (post-review, commit 8e56d23): exactly-once redesign + replica-task epoch cancellation
Attach-under-write stress testing surfaced three defects; all fixed on this branch before merge:
REPLICAOFleaked the previous replica task (pre-existing, found via a shards=1 control). A newREPLICAOFspawned a fresh apply task without stopping the old one, andNO ONEonly flipped the role — each detach/re-attach cycle stacked one more live applier (replica INCR counters +25-35% vs master). Replica tasks now carry a process-global epoch ticket;StartReplication/PromoteToMasterbump the generation, and superseded tasks exit before reconnecting, before snapshot load, and before applying any chunk.cut = shard offset at body capture; every live record carries its per-shardend_offset; delivery requiresend_offset > cut. All N shards now use the samePrepareReplicaSyncarm.ReplicaLiveFanoutarm: per-shard wire order == offset order by construction.New regressions:
attach_under_write_no_double_apply(4-shard + single-shard control; red before the fixes, +35%/+24% divergence),same_key_write_order_parity(12 conns APPEND-race 32 shared keys; byte-exact parity). Test-harness hardening from the VM sweep:replication_hardeningnow honorsMOON_BIN;aof_multidb_kill9'swait_readyretries on RESET during the bootstrap→per-shard listener handover (VM-timing flake, reproduced with main's binary).Round-2 gates (all green): multishard 9/9, streaming 7/7, hardening 5/5, graph 3/3, kill-9 4/4 — macOS AND Linux VM; lib 4115 (monoio) + 3316 (tokio); clippy
-D warnings×2; VM A/B bench (2 reps): s4 no-replica branch ≥ main on every metric (e.g. SET P16 2.67M vs 2.17-2.47M), s1 replica-attached branch ≥ main (SET P1 380-390k vs 328-365k), s4 replica-attached 2.1-2.3M SET P16 with exact DBSIZE parity in every run.Also included: R3 (RFC 2A + 4B) — tokio error path + docs
runtime-tokiomaster now answers PSYNC with-ERR PSYNC requires runtime-monoio on the master (this build runs runtime-tokio)instead of a generic unknown-command reply — verified live against real tokio binaries at shards=1 (handler_single) and shards=2 (handler_sharded).docs/guides/clustering.mddeployment shape,README.mdreplication bullets (also fixes the stale claim that replicas could run--shards N),docs/PRODUCTION-CONTRACT.mdREPL-MULTISHARD-01 + WAIT-01 flipped to ✅.Summary by CodeRabbit
WAITbehavior, and consistent degradation from partial to full resync.WAITsemantics.