fix(replication): replicas apply streamed SWAPDB, emitted exactly once (#386) - #442
Merged
Conversation
#386) Two stacked defects made SWAPDB silently diverge master and replica: 1. Replica no-op: `apply_local` had no SWAPDB intercept — the record fell through to generic dispatch, which hard-errors ("SWAPDB must be issued at the connection handler level"), and `warn_on_error` only logs. Every streamed SWAPDB no-op'd; the replica served pre-swap data for BOTH databases until a full resync. 2. Wire multiplicity: a multi-shard master emitted the record once per REMOTE shard leg (SwapDb SPSC arm -> wal_append_and_fanout) and never for the coordinator's own leg. Today's replica applies the merged wire as ONE stream, record by record — N-1 emissions swap N-1 times, a net NO-OP whenever N-1 is even (e.g. --shards 3). Correctness depended on the master's shard-count parity. The wire contract is now: exactly ONE SWAPDB record per client SWAPDB. - coordinator.rs (`coordinate_swapdb`): emit the replication record via `record_local_write_global` AFTER the durability gate and the local swap — an aborted SWAPDB can never reach replicas. debug_assert relaxed to >= 1 (monoio routes all shard counts here). - spsc_handler.rs (SwapDb arm): remote legs keep their per-shard WAL v3 + AOF writes (per-shard crash recovery replays each shard's own record) but no longer touch the replication backlog/offset/fanout. - apply.rs: SWAPDB intercept before generic dispatch — same slice-split swap as the WAL replay intercept; out-of-range / same-index / malformed args skip with a warn instead of poisoning the stream (a replica with fewer --databases must survive the record). - handler_single.rs (legacy non-sharded tokio listener, no production callers): emits the record after its swap for contract consistency. When #406 lands per-shard demuxed multi-shard replicas, this must flip to per-shard emission + per-stream apply (noted at all three sites). Tests (red/green): tests/replication_swapdb.rs — master at shards=1/3/4 with a live replica; shards=3 (even remote-leg count) pins the multiplicity defect, shards=4 pins the coordinator-leg emission; plus post-swap write replication. `#[ignore]`d like the other replication suites: PSYNC-as-master is monoio-only ("-ERR PSYNC requires runtime-monoio on the master"), so they run explicitly against a monoio release binary, never in the CI tokio job. Unit tests for `apply_swapdb` (both runtimes) cover swap, integer args, reversed order, out-of-range, same-index, malformed args. Gates: crash_matrix_per_shard_aof (SWAPDB kill-9 durability, #133) green; full monoio release suite green (one pre-existing client-tracking flake, fails identically on pristine main); clippy both feature sets; fmt. No hot-path code touched (SWAPDB paths only). Fixes #386 author: Tin Dang
📝 WalkthroughWalkthrough
ChangesSWAPDB replication
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related PRs
Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant Client
participant SWAPDBHandler
participant ShardCoordinator
participant AOF
participant GlobalReplication
participant ReplicaApply
participant ReplicaSlices
Client->>SWAPDBHandler: Execute SWAPDB
SWAPDBHandler->>AOF: Enqueue serialized WAL frame
SWAPDBHandler->>GlobalReplication: Record after durability and local swap
ShardCoordinator->>GlobalReplication: Emit one SWAPDB record
GlobalReplication->>ReplicaApply: Deliver replicated record
ReplicaApply->>ReplicaSlices: Swap local database slices
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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 |
PR Summary by QodoFix replica SWAPDB apply and enforce exactly-once SWAPDB replication record (#386)
AI Description
Diagram
High-Level Assessment
Files changed (6)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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 `@CHANGELOG.md`:
- Around line 10-24: Move the SWAPDB changelog bullet so it follows the complete
existing changelog item that begins before the current entry and ends after it.
Preserve the SWAPDB text unchanged and ensure the preceding c10k item remains a
single complete Markdown bullet.
In `@src/replication/apply.rs`:
- Around line 806-831: The apply_swapdb function currently ignores arguments
beyond the first two, allowing malformed SWAPDB records to execute. Require
args.len() == 2 before parsing and swapping, otherwise follow the existing
warning-and-skip path; add a test covering an extra-argument record and
confirming databases remain unchanged.
In `@src/shard/coordinator.rs`:
- Around line 2978-2989: Delay the coordinator’s record_local_write_global call
in src/shard/coordinator.rs:2978-2989 until every remote leg confirms successful
application and required durability, retaining serialized for that final
emission; update src/shard/spsc_handler.rs:2479-2499 to propagate
send_append_bounded_blocking failures to the coordinator and prevent SWAPDB or
success acknowledgement when the remote AOF append fails.
🪄 Autofix
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 Plus
Run ID: cec22b47-acdc-43f6-a0ae-2efe45f59060
📒 Files selected for processing (6)
CHANGELOG.mdsrc/replication/apply.rssrc/server/conn/handler_single.rssrc/shard/coordinator.rssrc/shard/spsc_handler.rstests/replication_swapdb.rs
Comment on lines
+10
to
+24
| - **Replicas now apply streamed `SWAPDB` (#386), and the record reaches the | ||
| wire exactly once per client call.** Two stacked defects: (1) the replica's | ||
| apply path had no SWAPDB intercept — generic dispatch hard-errors ("must be | ||
| issued at the connection handler level") and the error was only logged, so | ||
| every streamed SWAPDB silently no-op'd and the replica served pre-swap data | ||
| for both databases until a full resync; (2) a multi-shard master emitted the | ||
| record once per REMOTE shard leg and never for the coordinator's own leg — | ||
| against today's single merged replica stream that means N−1 swaps, a net | ||
| no-op whenever N−1 is even (e.g. `--shards 3`). The coordinator now emits | ||
| the replication record exactly once, after the durability gate and the | ||
| local swap (an aborted SWAPDB can never ship to replicas); remote SPSC legs | ||
| keep their per-shard AOF/WAL writes but stay off the replication plane; the | ||
| tokio single-shard handler emits it too. Replicas apply it with the same | ||
| slice-split swap as WAL replay, skipping (with a warning) indexes outside | ||
| their own `--databases` range instead of poisoning the stream. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Move the SWAPDB entry after the existing changelog item.
Line 10 starts a new bullet before the item that continues at Line 25 ends. Markdown attaches the c10k text to the new SWAPDB bullet and leaves the prior item incomplete. Move lines 10-24 after the complete existing item.
🤖 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 `@CHANGELOG.md` around lines 10 - 24, Move the SWAPDB changelog bullet so it
follows the complete existing changelog item that begins before the current
entry and ends after it. Preserve the SWAPDB text unchanged and ensure the
preceding c10k item remains a single complete Markdown bullet.
Comment on lines
+806
to
+831
| fn apply_swapdb(cmd: &[u8], args: &[Frame], databases: &mut [crate::storage::Database]) { | ||
| let parse_idx = |f: &Frame| match f { | ||
| Frame::BulkString(b) => std::str::from_utf8(b).ok()?.parse::<usize>().ok(), | ||
| Frame::Integer(n) => usize::try_from(*n).ok(), | ||
| _ => None, | ||
| }; | ||
| match ( | ||
| args.first().and_then(parse_idx), | ||
| args.get(1).and_then(parse_idx), | ||
| ) { | ||
| (Some(a), Some(b)) if a != b && a < databases.len() && b < databases.len() => { | ||
| let (lo, hi) = if a < b { (a, b) } else { (b, a) }; | ||
| // Split the slice to get two non-overlapping mutable references. | ||
| let (left, right) = databases.split_at_mut(lo + 1); | ||
| std::mem::swap(&mut left[lo], &mut right[hi - lo - 1]); | ||
| } | ||
| (Some(a), Some(b)) if a == b => {} // same-index: no-op, matches Redis | ||
| _ => { | ||
| tracing::warn!( | ||
| "replication apply: skipping {} with unusable args (out of range for {} local dbs)", | ||
| String::from_utf8_lossy(cmd), | ||
| databases.len() | ||
| ); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Reject extra SWAPDB arguments.
Line 812 parses only the first two arguments. A malformed record such as SWAPDB 0 1 extra still swaps databases. Require args.len() == 2 before parsing. Warn and skip all other argument counts. Add a test for the extra-argument case.
🤖 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/apply.rs` around lines 806 - 831, The apply_swapdb function
currently ignores arguments beyond the first two, allowing malformed SWAPDB
records to execute. Require args.len() == 2 before parsing and swapping,
otherwise follow the existing warning-and-skip path; add a test covering an
extra-argument record and confirming databases remain unchanged.
Comment on lines
+2978
to
+2989
|
|
||
| // #386 — replication plane, exactly once per client SWAPDB. Today's | ||
| // replica applies the merged wire as ONE stream, so the record must | ||
| // appear on it exactly once: the coordinator emits it here, AFTER | ||
| // the durability gate (an abort above never reaches this line, so a | ||
| // failed SWAPDB can never ship to replicas) and after the local | ||
| // swap; the remote legs' SPSC arms write AOF/WAL only. Safe on both | ||
| // runtimes: this runs on the shard's own OS thread (monoio shard | ||
| // thread / tokio per-shard LocalSet), whose event loop drains | ||
| // `self_msg`. When #406 lands per-shard demuxed replicas this must | ||
| // flip to per-shard emission. | ||
| crate::replication::state::record_local_write_global(my_shard, serialized); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Confirm every shard leg before global replication emission.
The coordinator records SWAPDB before remote legs report completion. A remote leg also discards a failed AOF append and still sends its success acknowledgement. A closed reply channel can return an error after the replica receives the global swap. An AOF enqueue failure can return +OK with no recoverable remote record.
src/shard/coordinator.rs#L2978-L2989: retainserializedand callrecord_local_write_globalonly after every remote leg confirms successful application and required durability.src/shard/spsc_handler.rs#L2479-L2499: propagatesend_append_bounded_blockingfailure to the coordinator. Do not swap or acknowledge success when the remote AOF append fails.
📍 Affects 2 files
src/shard/coordinator.rs#L2978-L2989(this comment)src/shard/spsc_handler.rs#L2479-L2499
🤖 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/coordinator.rs` around lines 2978 - 2989, Delay the coordinator’s
record_local_write_global call in src/shard/coordinator.rs:2978-2989 until every
remote leg confirms successful application and required durability, retaining
serialized for that final emission; update src/shard/spsc_handler.rs:2479-2499
to propagate send_append_bounded_blocking failures to the coordinator and
prevent SWAPDB or success acknowledgement when the remote AOF append fails.
Code Review by Qodo
1. Tokio calls shard-only fanout
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+910
to
+912
| crate::replication::state::record_local_write_global( | ||
| 0, serialized, | ||
| ); |
There was a problem hiding this comment.
3. Tokio calls shard-only fanout 🐞 Bug ≡ Correctness
handler_single calls replication::state::record_local_write_global, but that function explicitly requires running on the shard OS thread because it pushes to the shard thread-local self_msg queue; calling it from the tokio connection handler can cause the SWAPDB replication record to never be fanned out (and can also corrupt replication offsets). This breaks the stated “exactly-once on the wire” contract whenever handler_single is exercised.
Agent Prompt
## Issue description
`src/server/conn/handler_single.rs` invokes `crate::replication::state::record_local_write_global(...)` from a **tokio** connection handler. That helper pushes to `crate::shard::self_msg` (thread-local queue) and is documented as **shard-thread-only**; tokio tasks must not push there. This means the replication fanout for SWAPDB can be silently lost.
## Issue Context
- `record_local_write_global` explicitly states the caller must be on the shard OS thread and warns that tokio tasks must not push to `self_msg`.
- `shard::self_msg` module docs reiterate the same constraint.
- The PR added this call specifically for SWAPDB replication plane emission.
## Fix Focus Areas
- src/server/conn/handler_single.rs[905-913]
- src/replication/state.rs[473-507]
- src/shard/self_msg.rs[26-34]
## Suggested fix approach
- **Do not call** `record_local_write_global` from `handler_single`.
- Either:
1) Implement/use a **tokio-safe** replication emission path for the single-thread handler (e.g., append to the replication backlog + fanout via a tokio-owned sender list / channel that is actually drained in this runtime), or
2) If `handler_single` is truly non-production / non-replicating, remove the SWAPDB replication emission and document that this handler does not support replication-plane emission.
Acceptance criteria:
- No shard-thread-only (`self_msg`) APIs are invoked from tokio handler code.
- SWAPDB replication emission from this handler is either correct (delivered) or intentionally absent with explicit documentation.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+2494
to
+2497
| 0, | ||
| 0, | ||
| bytes::Bytes::copy_from_slice(&serialized), | ||
| &mut aof_budget, |
There was a problem hiding this comment.
4. Unneeded bytes copy in swapdb 🐞 Bug ➹ Performance
The SWAPDB SPSC arm allocates and copies the already-owned Bytes from serialize_command via Bytes::copy_from_slice(&serialized) before sending it to send_append_bounded_blocking, adding avoidable heap work on each SWAPDB remote leg. This should pass serialized.clone() (cheap) or move serialized when possible.
Agent Prompt
## Issue description
In the SWAPDB SPSC arm, `serialized` is already a `bytes::Bytes` (from `aof::serialize_command`). The code currently does `Bytes::copy_from_slice(&serialized)` before calling `send_append_bounded_blocking`, which allocates and copies the payload unnecessarily.
## Issue Context
- `aof::serialize_command` returns `Bytes`.
- `AofWriterPool::send_append_bounded_blocking` takes `Bytes` by value.
- Therefore, `serialized.clone()` is the correct low-cost way to pass ownership.
## Fix Focus Areas
- src/shard/spsc_handler.rs[2487-2499]
- src/persistence/aof/mod.rs[509-514]
- src/persistence/aof/pool.rs[498-505]
## Suggested fix approach
- Replace `bytes::Bytes::copy_from_slice(&serialized)` with `serialized.clone()` (or move `serialized` if no longer needed after WAL append).
- Keep WAL append using `&serialized` as-is.
Result:
- Eliminates an avoidable allocation/copy on SWAPDB remote legs.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
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
1 participant
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.
Fixes #386.
Two stacked defects
Replica no-op:
apply_localhad no SWAPDB intercept — the record fell through to generic dispatch, which hard-errors ("SWAPDB must be issued at the connection handler level"), andwarn_on_erroronly logs. Every streamed SWAPDB silently no-op'd; the replica served pre-swap data for BOTH databases until a full resync.Wire multiplicity (found during design): a multi-shard master emitted the record once per REMOTE shard leg (SwapDb SPSC arm →
wal_append_and_fanout) and never for the coordinator's own leg. Today's replica applies the merged wire as ONE stream, record by record — N−1 emissions swap N−1 times, a net NO-OP whenever N−1 is even (e.g.--shards 3). Correctness depended on the master's shard-count parity.The fix — exactly-once wire contract
coordinate_swapdb): emit the replication record viarecord_local_write_globalAFTER the durability gate and the local swap — an aborted SWAPDB can never reach replicas.When #406 lands per-shard demuxed multi-shard replicas, this must flip to per-shard emission + per-stream apply (noted at all three sites).
Tests (red/green TDD)
tests/replication_swapdb.rs— real master (shards=1/3/4) + real replica. All three failed before the fix (db0 k0=\"before-0\"— replica kept pre-swap state), all green after. shards=3 is the load-bearing case: two remote legs = even swap count, so a replica-apply-only fix nets to no-op and the test catches the multiplicity defect. shards=4 pins the coordinator-leg emission. Plus post-swap write replication, and 3 unit tests forapply_swapdb(swap, integer args, reversed order, out-of-range, same-index, malformed).The integration tests are
#[ignore]d like the other replication suites: PSYNC-as-master is monoio-only (-ERR PSYNC requires runtime-monoio on the master), so they run explicitly against a monoio release binary and can never pass in the CI tokio job.Gates
crash_matrix_per_shard_aof4/4 incl.crash_133_swapdb_multishard_durability_after_sigkillclient_tracking_invalidationflake — fails identically on pristine main, A/B verified)runtime-tokio,jemalloc; clippy-D warningsboth feature sets; fmtSummary by CodeRabbit
Bug Fixes
SWAPDBreplication across single- and multi-shard deployments.Tests