feat(replication): R1 real WAIT/ACK + consistency/durability fixes (silent replica drop, SELECT leak, AOF db attribution)#282
Merged
Conversation
…reader, WAIT wire Task #19, the v0.7 R1 milestone item: WAIT/ACK was dead in production. Three gaps closed (the fourth, per-shard ack floors, keeps the existing Vec<AtomicU64> shape and rides the R2 multi-shard redesign): 1. Replica ACK send (replica.rs, both runtimes): the replication socket is split (monoio::io::Splitable / tokio into_split); a dedicated 1s ticker task owns the write half and sends REPLCONF ACK <master_repl_offset> — Redis's replicationCron cadence, which also serves as an idle keepalive for master-side lag detection. A timeout-wrapped read was rejected by design: cancelling an in-flight io_uring read whose CQE already completed DISCARDS those bytes — silent replication-stream corruption. The read/apply loop is unchanged, extracted as stream_commands_read_loop; the ticker stops via a per-connection flag when the read loop exits, and the write half drops with it. 2. Master ACK read (master.rs): drain_replica_inline_single_shard splits the hijacked PSYNC socket; a same-thread reader task (ack_read_loop) parses REPLCONF ACK frames and records them into the replica's ack_offsets/last_ack_time. fetch_max — a reordered or duplicate ACK can never regress the offset. Parsing is a dedicated drain_ack_offsets over protocol::parse: the shared replication drainer (drain_replicated_commands) deliberately DROPS REPLCONF as chatter, which is exactly the frame this loop exists to read (first implementation reused it and swallowed every ACK — caught by the red e2e staying red). Malformed bytes close the connection loudly; the replica then reconnects with a fresh resync. 3. WAIT wire (handler_monoio + handler_sharded dispatch): WAIT answered ':0' unconditionally from the synchronous generic dispatch table on the monoio runtime (only handler_single ever called wait_for_replicas). New connection-layer try_handle_wait intercept awaits wait_for_replicas (10ms poll, early exit). Redis parity: timeout 0 = block until satisfied (capped at one year — the poll loop is cooperative so the shard thread is never starved). The dispatch-table arm remains as a truthful fallback for paths without repl_state (no replicas can be attached there). Red/green: new e2e wait_returns_acked_replica_count — WAIT with no replicas returns 0 fast; WAIT 1 after a write returns 1 within the ACK cadence; WAIT 2 with one replica times out reporting 1. RED before (returned :0), GREEN after; the intermediate drainer-reuse bug kept it red until the dedicated parser landed. Verification: lib 4111 green, replication_streaming 6/6 (incl. new test), fmt clean; clippy + tokio matrix + consistency/durability/bench gates run before PR (see PR description). author: Tin Dang
Task #35. The user-mandated consistency/durability gates for the R1 WAIT/ACK work (multi-db load parity + kill-9 recovery, A/B'd against main) exposed three PRE-EXISTING data-integrity bugs. All three reproduced on main with the same harness; none are R1 regressions. 1. Replication silently dropped records for lagging replicas. wal_append_and_fanout / ReplicaLiveFanout used try_send and SKIPPED the record on a full channel — one pipelined burst left the replica with ~2k of 40k keys, master_link_status:up, offsets diverged forever (WAIT correctly reported 0, which is how R1 exposed it). Fix: fanout_send_or_kick — overflow sets a shared `kicked` flag (ShardMessage::RegisterReplica now carries it) and drops the shard's sender; the inline drain loop polls the flag (monoio::select! with a 250ms timer — the kick cannot arrive in-band on a full channel, and ReplicaInfo.shard_txs holds a sender clone so channel closure cannot signal it either) and closes the socket. The replica reconnects and resyncs from the backlog — Redis's output-buffer-limit policy: divergence becomes a loud, self-healing resync. Channel capacity 1024 -> 16384 records so ordinary bursts kick rarely. 2. Client SELECT commands leaked into BOTH persistence planes. SELECT is W-flagged for routing, so every handler AOF'd/replicated the literal client SELECT while bare writes carried no db context. Interleaved connections then corrupted db attribution downstream — conn A (db0) SET after conn B's SELECT 2 handshake replayed into db2. New metadata::is_persisted_write (is_write minus SELECT) gates all four handler persist legs; SELECT is connection-state only. 3. The AOF had no per-record db attribution at all. Kill-9 recovery of a 20k-db0 + 20k-db2 load restored 0/40000 (every key into db2 — the last literal client SELECT won). Fix mirrors Redis's aof_selected_db: AofMessage::{Append,AppendSync} carry the executing db; each writer task tracks the stream's current db and injects a `SELECT <db>` record exactly when it changes, in the same on-disk format (raw RESP for TopLevel, [lsn=0][len] framed for PerShard). Context resets per fresh incr segment (replay starts each segment at db0) and threads through every BGREWRITEAOF fold drain; drains that leave the context ambiguous set a usize::MAX force-emit sentinel — a redundant SELECT is harmless, a missing one is corruption. Zero-length fsync-barrier payloads never trigger nor observe a db switch. Replay engines already executed SELECT records; only the writer was blind. Red/green: - NEW tests/aof_multidb_kill9.rs (TopLevel shards=1 + PerShard shards=2): interleaved two-conn multi-db writes, kill -9, restart, exact per-db placement. RED before (db0=0, all keys in db2), GREEN after, on BOTH runtimes. - NEW replica_converges_under_interleaved_multidb_load (replication_streaming): 2x10k interleaved pipelined SETs across db0/db2 with a replica attached; exact DBSIZE parity + no cross-db leaks after settle. RED before (5116/10001, 5115/10000), GREEN after. - Writer-level unit test: Append db0 -> db2 -> db0 produces framed SELECT records at exactly the two switches. Verification: gate script 11/11 PASS (WAIT-under-load 1, 20000/20000 parity both dbs, AOF recovery 20000/20000, post-restart reconvergence); replication e2e matrix 17/17 (streaming 7, graph 3, hardening 5, aof_multidb 2); crash_matrix_per_shard_aof 3/3 + crash_matrix_per_shard_bgrewriteaof 2/2 (rewrite-fold canary); lib 4114 monoio / 3315 tokio; clippy -D warnings both matrices; fmt clean. 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? |
|
Warning Review limit reached
Next review available in: 31 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 (1)
📝 WalkthroughWalkthroughThe change adds database-aware AOF persistence with synthetic ChangesAOF database attribution
Replication R1 ACK and fan-out behavior
Estimated code review effort: 5 (Critical) | ~120 minutes 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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)
src/server/conn/shared.rs (1)
238-249: 🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy liftMULTI/EXEC still persists queued
SELECTand mis-tags later writes.
aof_bytesis still gated onmetadata::is_write(cmd), so a queuedSELECTlands inaof_entriesand reaches AOF/replication.persist_txn_aof(..., db)takes one db for the whole body, butSELECTmutates the localselected, so later writes can be replayed under the wrong db.Switch this path to
is_persisted_writeand either forbid db changes insideMULTIor thread the db per entry.🤖 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/shared.rs` around lines 238 - 249, Update the sharded MULTI/EXEC AOF handling around aof_bytes and persist_txn_aof: gate serialization with is_persisted_write instead of metadata::is_write so queued SELECT is excluded, then prevent database changes within MULTI or propagate the selected database per transaction entry so later writes persist under the correct database.
🤖 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/server/conn/handler_single.rs`:
- Around line 1391-1399: Update execute_transaction and its txn_aof_entries
representation to retain the active database for each queued command’s AOF
entry, including changes around SELECT handling and write recording. When
merging entries in the EXEC path, use each entry’s captured database rather than
the final conn.selected_db, while preserving the existing exec_resp_idx
association.
---
Outside diff comments:
In `@src/server/conn/shared.rs`:
- Around line 238-249: Update the sharded MULTI/EXEC AOF handling around
aof_bytes and persist_txn_aof: gate serialization with is_persisted_write
instead of metadata::is_write so queued SELECT is excluded, then prevent
database changes within MULTI or propagate the selected database per transaction
entry so later writes persist under the correct database.
🪄 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: 5d9e9260-c608-47b9-921d-15550ad39a3e
📒 Files selected for processing (26)
CHANGELOG.mdsrc/command/metadata.rssrc/command/mod.rssrc/persistence/aof/mod.rssrc/persistence/aof/pool.rssrc/persistence/aof/rewrite.rssrc/persistence/aof/writer_task.rssrc/persistence/aof_manifest/shard_replay.rssrc/replication/master.rssrc/replication/replica.rssrc/server/conn/blocking.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/write.rssrc/server/conn/handler_sharded/dispatch.rssrc/server/conn/handler_sharded/mod.rssrc/server/conn/handler_sharded/write.rssrc/server/conn/handler_single.rssrc/server/conn/shared.rssrc/shard/coordinator.rssrc/shard/dispatch.rssrc/shard/event_loop.rssrc/shard/spsc_handler.rstests/aof_multidb_kill9.rstests/replication_streaming.rstests/wal_group_commit.rs
PR #282 review (CodeRabbit, Major): the txn executors collapsed every body entry to ONE db at EXEC time and still persisted the literal queued SELECT (they gated on is_write — missed by the top-level is_persisted_write sweep). Deterministic corruption: `MULTI; SELECT 2; SET t2; EXEC` then a plain db0 write — the literal SELECT 2 shifts the replay/replica stream to db2, but the next db0 record is tagged 0 == the writer's tracker 0, so no corrective SELECT is injected: every post-EXEC db0 write recovers into db2 (and applies into the replica's db2). RED-proven by the new e2e before the fix. Per-executor semantics (they genuinely differ): - execute_transaction_sharded re-dispatches each body command against the CURRENT `selected`, so a queued SELECT really redirects the commands after it. aof_entries is now Vec<(usize, Bytes)> with the db captured per entry BEFORE dispatch; persist_txn_aof and the monoio EXEC replication leg (record_local_write_db) consume it per entry, as does the TxnExecute owner-routing arm's wal_append_and_fanout loop. - execute_transaction (handler_single) holds ONE guard on the db selected at EXEC time — every body write physically lands there even if a queued SELECT mutates conn.selected_db mid-body. The caller now captures `txn_db` BEFORE the call and attributes all entries to it (the post-EXEC conn.selected_db is the wrong answer there). - Both executors now use is_persisted_write: a queued literal SELECT is never persisted on any path. Red/green: new e2e aof_txn_select_inside_multi_kill9_{toplevel, per_shard} — MULTI body with a queued SELECT, post-EXEC db0 probe, kill -9, restart; recovery placement must equal memory placement. RED before (post lands in db2), GREEN after, on BOTH runtimes. Regression: aof_multidb_kill9 4/4 both runtimes, replication_streaming 7/7, replication_hardening 5/5, crash_matrix_per_shard_aof 3/3, crash_matrix_per_shard_bgrewriteaof 2/2, lib 4114, clippy -D warnings both matrices, fmt clean. Known residue (unchanged semantics, noted for R2): in-txn SELECT does not persist to the connection after EXEC on the sharded paths (`selected` mutations are local to the executor), and OrderedAcrossShards merged entries still replay per-entry at db0. author: Tin Dang
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 `@tests/aof_multidb_kill9.rs`:
- Around line 217-233: Update the memory-baseline checks around mem_t2_db0 and
mem_t2_db2 to require t2 exists in exactly one database. Add an assertion that
exactly one GET result is non-missing, while preserving the existing
per-database reads and later recovery comparisons.
🪄 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: 4d2d4e18-845a-4e1d-8cd5-0af6c3c51040
📒 Files selected for processing (5)
src/server/conn/handler_monoio/write.rssrc/server/conn/handler_single.rssrc/server/conn/shared.rssrc/shard/spsc_handler.rstests/aof_multidb_kill9.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- src/server/conn/handler_single.rs
- src/shard/spsc_handler.rs
PR #282 review (Minor): the memory baseline for t2 could be `$-1` in BOTH dbs if the queued `SET t2` silently failed, letting every recovery assertion pass vacuously. Assert exactly-one-db placement before the kill so the test can never green on a no-op transaction. author: Tin Dang
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
Two commits completing v0.7 R1 (task #19) plus the three pre-existing consistency/durability P0s the new gates caught (task #35):
Commit 1 — R1: real WAIT/ACK plumbing
WAIT was dead in production (returned
:0unconditionally on monoio). Three legs:monoio::io::Splitable/ tokiointo_split); a 1s ticker task owns the write half and sendsREPLCONF ACK <offset>(replicationCron cadence + idle keepalive). A timeout-wrapped read was rejected by design: cancelling an in-flight io_uring read whose CQE completed discards bytes → silent stream corruption.ack_read_loopparses ACKs via a dedicateddrain_ack_offsetsparser (the shared replication drainer deliberately drops REPLCONF as chatter — reusing it swallowed every ACK, caught by the red e2e).fetch_maxso reordered/duplicate ACKs never regress.try_handle_waitintercept (monoio + tokio sharded) awaitswait_for_replicas;timeout 0= block (capped 1y). Per-shard ack floors ride R2.Commit 2 — consistency/durability fixes (all A/B-reproduced on main; none are R1 regressions)
try_sendskipped records on a full channel — one pipelined burst left a replica with ~2k of 40k keys,master_link_status:up, diverged forever. Now overflow kicks the replica (sharedkickedflag → drain loop closes the socket → PSYNC resync) — Redis's output-buffer-limit policy. Channel 1024→16384.SELECTpersisted as a command (it's W-flagged for routing): poisoned the db context of BOTH persistence planes for interleaved connections. Newmetadata::is_persisted_writeexcludes it at all four handler persist gates.AofMessage, tracks the stream's current db, and injectsSELECT <db>records exactly on change (Redisaof_selected_db), across all four writer loci + BGREWRITEAOF fold drains.Verification (user-mandated consistency/durability/bench gates)
replica_converges_under_interleaved_multidb_load+wait_returns_acked_replica_count), graph 3, hardening 5, newaof_multidb_kill92. Tokio: lib 3315,replication_test5/5,aof_multidb_kill92/2.crash_matrix_per_shard_aof3/3,crash_matrix_per_shard_bgrewriteaof2/2.Known follow-ups
Summary by CodeRabbit
New Features
WAIT, returning the number of replicas that acknowledge a requested replication offset.Bug Fixes
SELECTcommands are no longer incorrectly persisted or replicated.