feat(replication): MQ plane — live stream + PSYNC registry blob + RDB Stream codec data-loss fix (Wave B stage 2b, task #34)#294
Merged
Conversation
…loss) The FULLRESYNC RDB codec (persistence::redis_rdb, distinct from the persistence::rdb SAVE/BGSAVE codec) serialized Stream values as a placeholder "__stream__:<len>" string — entries, last_id, consumer groups, PEL state, durable flag, and max_delivery_count were all silently discarded on every replica full resync. A durable MQ queue snapshotted this way would come back empty on the replica. Add a full RDB_TYPE_STREAM_MOON (0xC8) codec, ported from persistence::rdb's battle-tested Stream serializer: entries, last_id, consumer groups (PEL + per-consumer pending), durable flag, and max_delivery_count round-trip losslessly. Every new count field is bounds-checked via check_alloc_bound before allocating, matching the existing DoS-avoidance discipline in this module. Also adds the MOON_AUX_MQ_REGISTRY aux-field constant used by the Wave B stage 2b MQ replication work landing in subsequent commits. author: Tin Dang
Preparatory plumbing for Wave B stage 2b MQ replication: - src/mq/wal.rs: five synthetic MQ._REPL.CREATE/PUSH/POP/ACK/TRIGGER pseudo-command names plus is_mq_replay_command(), used to frame the existing encode_mq_*/decode_mq_* WAL payload bytes on the wire and to route them on the replica side (never dispatched to real clients). - src/mq/trigger.rs: TriggerRegistry::iter(), needed to snapshot every registered trigger for the FULLRESYNC registry blob landing in a later commit. - src/shard/shared_databases.rs: generalize apply_mq_create/push/pop/ ack/trigger over a new MqApplyTarget trait (mq_databases_mut() / mq_durable_queue_registry_mut() / mq_trigger_registry_mut()), implemented for both ShardSliceInit (boot-time WAL replay) and ShardSlice (live replica apply, added next). One apply engine now serves both boot recovery and replica replication instead of two. author: Tin Dang
… leg Wave B stage 2b, replica-consuming half. Two pieces: 1. Live-record apply (src/replication/apply.rs): apply_local routes any MQ._REPL.* record to a new apply_mq(), which decodes the single bulk-string payload with the existing decode_mq_* functions and dispatches to the SAME apply_mq_* engine (shard::shared_databases, generalized last commit) boot-time WAL replay uses. A replica never fires MQ.TRIGGER callbacks on apply — registrations are stored as opaque data, matching boot-time replay. 2. FULLRESYNC snapshot leg (src/replication/mq_sync.rs, new): a per-shard MOON_AUX_MQ_REGISTRY blob carries the durable-queue registry and trigger registry — shard-level bookkeeping that lives outside the keyspace (Stream CONTENT rides the ordinary RDB body, covered by the redis_rdb.rs Stream codec fix). Wired at both master capture sites (single-shard PSYNC in master.rs, and the merged multi-shard path via a new PreparedShardSync.mq_blob field populated in spsc_handler.rs) and installed additively into every replica shard by load_snapshot (apply.rs), mirroring graph_sync::install_graph_store_many's precedent exactly: an EMPTY blob list (pre-MQ-replication master) warns-and-keeps local state; a non-empty list is authoritative and replaces it. author: Tin Dang
Emits MQ.* effect records to attached replicas at the moment they're durably WAL-logged on the owner shard, closing the plane gap warn_unreplicated_plane previously fail-loud-warned about for MQ.* (the WS.* half of that warning is unaffected — a separate agent tracks retiring it once WS replicates). - src/replication/state.rs: record_local_write_global / record_local_write_db_global — ctx-free counterparts of record_local_write(_db) that resolve the same per-process ReplicationState Arc via admin::metrics_setup::get_global_repl_state_arc(), needed because shard::mq_exec runs on the shard's own OS thread without a ConnectionContext in scope. Also adds ReplicationState::num_shards(). - src/shard/mq_exec.rs: replicate_mq_record(), called from all five owner-shard handlers (CREATE/PUSH/POP/ACK/TRIGGER) right where each already builds its WAL payload — ships the IDENTICAL encoded bytes as one of the MQ._REPL.* pseudo-commands, gated single-shard-only (num_shards == 1), the same posture graph's own live path takes for its multi-shard gap. monoio-only; a no-op under runtime-tokio. - src/server/conn/handler_monoio/txn.rs: MQ.PUBLISH's TXN-materialization leg replicates the same way at commit, on the connection's own thread (which — since owner == ctx.shard_id in this leg — IS the owning shard's thread). - src/server/conn/handler_monoio/write.rs: retire the MQ half of warn_unreplicated_plane's call site now that MQ replicates in the single-shard configuration; the WS.CREATE/DROP call site is untouched. A multi-shard master still does not live-stream MQ writes (durability is unaffected — WAL + FULLRESYNC still cover it); tracked as a known follow-up alongside graph's own multi-shard live-stream gap. author: Tin Dang
tests/replication_mq.rs, modeled on tests/replication_graph.rs + tests/replication_planes.rs (single-shard master scope, matching graph's own precedent): - replica_syncs_mq_live_stream: MQ.CREATE/PUSH/ACK on the master reach an already-attached replica, verified via MQ.POP on the replica side (proves both registry state and message content replicated, not just the raw key) plus a stream-health assertion (zero reconnects). - replica_backfills_mq_from_snapshot: a replica attaching to a master that already has MQ state backfills it from PSYNC FULLRESYNC (the redis_rdb.rs Stream codec fix is what makes this pass), then a live write after the snapshot boundary still lands. - multi_shard_live_mq_write_not_streamed: pins the documented num_shards==1 live-stream gate down with an explicit RED-if-lifted test rather than leaving the limitation untested. Two new cargo-fuzz targets (wired into both the PR and nightly matrices in .github/workflows/fuzz.yml, per the project's "any new parser/decoder needs a fuzz target" rule): - mq_registry_blob: fuzzes install_mq_registry_many (the new MOON_AUX_MQ_REGISTRY blob decoder) on a throwaway per-iteration ShardSlice, mirroring mq_wal_record.rs's coverage of the sibling WAL op-blob decoders. - redis_rdb_load: fuzzes persistence::redis_rdb::load_rdb directly (the FULLRESYNC RDB loader, including the new RDB_TYPE_STREAM_MOON tag) — distinct from the existing rdb_load.rs target, which covers the separate persistence::rdb SAVE/BGSAVE codec. author: Tin Dang
author: Tin Dang
…dec fix Adversarial review of the mq-repl branch (verdict SHIP, P2 finding): the RDB_TYPE_STREAM_MOON fix changes the PSYNC wire format for durable Streams, so mixed-version pairs misbehave in both directions — old replica vs new master hard-fails FULLRESYNC (reconnect-loops until upgraded), new replica vs old master absorbs the legacy placeholder as a corrupted string. Document the upgrade-together requirement where operators will see it. author: Tin Dang
Rebase-composition tidy: with both the WS plane (PR #293) and this branch's MQ live stream shipped, the round-2 warn_unreplicated_plane fail-loud marker has no surviving half — the renamed warn_mq_unreplicated lost its last call site during the rebase, so the function is removed (dead code under -D warnings) and the doc comment in command/mq.rs plus the WS-plane CHANGELOG entry that still claimed "warn_mq_unreplicated survives for MQ" are corrected. author: Tin Dang
Rebase-composition fix: the replication_mq suite verified replica state by issuing MQ POP against the replica, and readonly enforcement (PR #292, merged after this branch was cut) now correctly rejects that with -READONLY — MQ POP claims PEL entries and advances last_delivered, a genuine write. The enforcement is right; the tests were exercising a hole it closed. Replica-side verification now uses read-only stream commands (XRANGE for content, XPENDING summary for the consumer-group PEL), which also makes the assertions STRONGER: the live-stream test now observes the master's MqPop claim land in the replica PEL (1 pending) and the MqAck drain it (0 pending) directly, instead of inferring both from what a replica-local POP re-surfaced. Both tests then add the operational failover proof — REPLICAOF NO ONE and a real MQ POP on the promoted node — which exercises the replicated durable flag, consumer-group state, and last_delivered cursor end-to-end the way an actual failover would. 3/3 green locally (macOS release, --include-ignored). 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? |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughAdds single-shard live replication for MQ mutations, full-fidelity stream RDB serialization, per-shard MQ registry FULLRESYNC blobs, replica-side replay and restore logic, fuzz targets, integration tests, and changelog coverage. ChangesMQ replication and snapshot support
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant MQExec
participant ReplicationState
participant ReplicaApply
Client->>MQExec: MQ mutation
MQExec->>ReplicationState: Serialize MQ._REPL.* record
ReplicationState->>ReplicaApply: Fan out live record
ReplicaApply->>ReplicaApply: Decode and apply MQ effect
Possibly related PRs
🚥 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.
Actionable comments posted: 5
🧹 Nitpick comments (1)
src/shard/mq_exec.rs (1)
279-287: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid constructing and copying an intermediate
Frameper MQ write.Lines 279-287 allocate an array and copy the payload for every replicated MQ mutation. Serialize the two RESP bulk strings directly into a preallocated buffer before calling
record_local_write_db_global.As per coding guidelines, shard hot paths should use preallocated buffers or borrowing rather than intermediate heap allocations.
🤖 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/mq_exec.rs` around lines 279 - 287, Update the MQ write path around record_local_write_db_global to avoid constructing a temporary Frame and copying payload into Bytes. Serialize the command name and payload RESP bulk strings directly into a preallocated buffer, then pass the serialized bytes to record_local_write_db_global while preserving the existing wire format and shard/db identifiers.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 `@src/persistence/redis_rdb.rs`:
- Around line 1773-1783: Update the truncation loop in the relevant test to
assert that read_rdb_entry rejects every truncated payload instead of discarding
its result. Check the returned Result after read_redis_string succeeds and fail
the test if decoding succeeds, while preserving the existing best-effort skip
for cuts where the key cannot be read.
In `@src/replication/apply.rs`:
- Around line 220-228: Update the MQ replay branch in the replication apply flow
to pass the already-clamped db_idx into apply_mq instead of relying on the
payload database index. Ensure all DB-scoped MQ appliers, including create,
PUSH, POP, and ACK handling, use this clamped target database while preserving
the existing apply_mq routing and behavior.
- Around line 473-503: Update the MQ POP replication flow around decode_mq_pop
and apply_mq_pop to carry each pending entry’s delivery timestamp and each
consumer’s seen timestamp in the payload, then pass and restore those values
instead of rebuilding them as zero. Update the corresponding POP
encoding/decoding and live application path while preserving existing stream,
claim, and DLQ data, and add a replay test that verifies non-zero timestamps
survive replication.
- Around line 455-528: Update the MQ replay handling around the command dispatch
and its caller so malformed payloads propagate an error instead of only logging
and continuing. Change the relevant apply path, including apply_local(), to
return failure on decode errors and prevent replica.rs from advancing
master_repl_offset; trigger the existing reconnect or full-resync behavior for
that failure while preserving successful command application.
- Around line 753-756: Update load_snapshot and its malformed MQ registry aux
blob error path to return the appropriate typed SnapshotError/MoonError variant
instead of anyhow::Result and anyhow::anyhow!. Preserve the existing
malformed-blob handling while propagating the typed error through callers,
retaining anyhow conversion only at the top-level binary boundary.
---
Nitpick comments:
In `@src/shard/mq_exec.rs`:
- Around line 279-287: Update the MQ write path around
record_local_write_db_global to avoid constructing a temporary Frame and copying
payload into Bytes. Serialize the command name and payload RESP bulk strings
directly into a preallocated buffer, then pass the serialized bytes to
record_local_write_db_global while preserving the existing wire format and
shard/db identifiers.
🪄 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: 08af3328-0e40-46cd-bb21-98ea5a52f321
📒 Files selected for processing (22)
.github/workflows/fuzz.ymlCHANGELOG.mdfuzz/Cargo.tomlfuzz/fuzz_targets/mq_registry_blob.rsfuzz/fuzz_targets/redis_rdb_load.rssrc/command/mq.rssrc/mq/trigger.rssrc/mq/wal.rssrc/persistence/redis_rdb.rssrc/replication/apply.rssrc/replication/master.rssrc/replication/mod.rssrc/replication/mq_sync.rssrc/replication/state.rssrc/server/conn/handler_monoio/ft.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_monoio/write.rssrc/shard/dispatch.rssrc/shard/mq_exec.rssrc/shard/shared_databases.rssrc/shard/spsc_handler.rstests/replication_mq.rs
💤 Files with no reviewable changes (1)
- src/server/conn/handler_monoio/ft.rs
Two CodeRabbit findings on PR #294: 1. apply_mq used the payload's db index unclamped — a replica configured with fewer logical dbs than its master would silently no-op the stream mutation (MqApplyTarget::mq_databases_mut().get_mut(oob) returns None): CREATE registered the queue but never created the stream, PUSH/POP/ACK were dropped. Now clamped with the same posture (and debug log) as the generic KV clamp in apply_local. Boot-time WAL replay is unaffected — a shard replays its own records, whose db indices are valid by construction. 2. The RDB stream truncation test discarded the decode result, proving only "no panic". Every strict prefix of the length-prefixed encoding must hit EOF, so it now asserts is_err() at every cut point. Gates: redis_rdb unit tests 30/30, replication_mq 3/3, clippy, fmt. 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
Final Wave B piece (task #34): the MQ plane now replicates — durable-queue
MQ.*mutations reach attached replicas via both the live stream and PSYNC FULLRESYNC, closing the lastwarn_unreplicated_planegap. Along the way this fixes a pre-existing FULLRESYNC data-loss bug: the PSYNC RDB codec serializedStreamvalues as a placeholder"__stream__:<len>"string, silently discarding all entries/PEL/consumer-group state for ANY stream (not just MQ).What's in here
persistence/redis_rdb.rs): fullRDB_TYPE_STREAM_MOON(0xC8) codec — entries,last_id, consumer groups (PEL + per-consumer pending),durableflag,max_delivery_count. Round-trip + truncation tests. ⚠ Operator note (CHANGELOG'd): upgrade masters and replicas together — mixed-version pairs misbehave in both directions (inherent to fixing a wire-format bug).shard/mq_exec.rs,replication/state.rs):replicate_mq_recordemits the SAME encoded payload bytes already built for the stage-2a WAL records as five syntheticMQ._REPL.*pseudo-commands (recognized ONLY inapply_localon the replication link — unreachable from client dispatch). Mutation → replicate → WAL append is one synchronous stretch with zero.await, preserving the offset-cut exactly-once invariant. Single-shard gate (num_shards == 1), same posture as graph's live path; pinned bymulti_shard_live_mq_write_not_streamed.replication/mq_sync.rs, new): per-shardMOON_AUX_MQ_REGISTRYblob carries the durable-queue + trigger registries (the bookkeeping OUTSIDE the keyspace; stream content rides the RDB body). Installed additively,graph_syncprecedent — absent aux = pre-MQ master, warn-and-keep.replication/apply.rs,shard/shared_databases.rs):apply_mqdecodes with the stage-2a codecs and dispatches to the SAMEapply_mq_*engine boot-time WAL replay uses (generalized overMqApplyTarget). Idempotent byStreamId— a record redelivered across the snapshot cut is a no-op. Replicas store trigger registrations as opaque data and never fire them.tests/replication_mq.rs(live stream w/ PEL-level XPENDING observation, snapshot backfill, multi-shard limitation pin — both replication tests end with aREPLICAOF NO ONEfailover-promotion proof), 2 new fuzz targets (mq_registry_blob,redis_rdb_load) wired into both CI matrices.Composition with #292/#293 (resolved during rebase)
warn_mq_unreplicatedlost its last call site → function retired, docs/CHANGELOG corrected.MQ POP, which readonly enforcement (fix(shard): reject client-issued WS/MQ/TEMPORAL writes on read-only replicas (Wave B) #292) now correctly rejects — rewritten to read-only observation (XRANGE/XPENDING) + promotion, which is strictly stronger.Gates
replicate_mq_recordlives in MQ execution sites only and early-returns when no replica is attached; same waiver basis as feat(replication): WS plane — shard-0-pinned deterministic WS.CREATE/DROP + live streaming + PSYNC registry blob (Wave B) #293).Closes the MQ half of task #34. Wave B complete.
Summary by CodeRabbit