feat(replication): v0.7 graph-plane replication — live GRAPH.* stream + PSYNC snapshot backfill#279
Merged
Merged
Conversation
… local-write fanout
The SPSC mesh is N·(N−1) with skip-self mapping, so a task on a shard's
own thread had NO producer to that shard. At --shards 1 the producer Vec
is EMPTY: the inline PSYNC task's RegisterReplica failed every attach
("shard 0 producer missing") and the replica fell into a 0.5s
reconnect/full-resync loop. R0/R0.5 tests stayed green because each
resync's RDB carried the latest keyspace + FT defs — data crawled across
via snapshot polling, masking the dead live stream entirely.
Repair, in four parts:
1. shard::self_msg — thread-local self-message queue (same pattern as
shard::slice), drained by drain_spsc_shared ahead of the SPSC
consumers and woken via the shard's own drain Notify. PSYNC
registration/unregistration and all replication fan-out on the owning
shard's thread route through it.
2. Local (same-shard) writes now feed the replication plane at all:
successful local writes push their wire bytes as ReplicateVerbatim
BEFORE any await (mutation + record are one synchronous stretch —
atomic w.r.t. the inline PSYNC snapshot capture). The drained message
does backlog + offset + replica fan-out together, so the AOF leg no
longer double-advances the offset (lsn = 0 when fan-out owns the
advance; per-shard order is append order, same contract as the
cross-shard legs).
3. ReplicationBacklog::new_at — the lazily-allocated backlog now seeds
its byte positions at the CURRENT shard offset. An unseeded (0-based)
backlog made every catch-up range read on a pre-written master fail
as "evicted", feeding the reconnect loop even after registration.
4. replication::state::fanout_hint_active() — process-global AtomicBool
(set on first replica attach, never cleared) gating all fan-out
serialization with one Relaxed load, so non-replicating servers pay
nothing on the hot path (S3.5a rationale).
Verified: master+replica manual soak shows zero reconnects / zero PSYNC
handler exits with KV + FT + graph live streaming; full lib suite,
replication_streaming/test/hardening, spsc_wake_floor green on monoio
and tokio feature sets.
author: Tin Dang
… + PSYNC snapshot backfill Closes the last unsynchronized data plane (task #25: "synchronize KV space and collections/indexes — vector, text, graph"). Live leg: graph mutations (GRAPH.* + Cypher writes) stream to replicas as their DETERMINISTIC, id-pinned WAL records (GRAPH.ADDNODE <g> <id> …) — the records gs.drain_wal() already produces for durability. Label and property-key ids are a stateless FNV-1a hash (label_to_id), so master and replica resolve identical ids with no interner-state sync. The replica applies records through the same GraphReplayCollector restart recovery uses (apply.rs intercept BEFORE generic dispatch — dispatch parses USER syntax and would re-allocate ids). Replay's edge/SETPROP resolution now also seeds node_map from write-buffer-resident nodes (external_id ↔ NodeKey is the KeyData bijection), so one-record-at-a- time streaming replay resolves endpoints applied by EARLIER records; a no-op during restart recovery. Snapshot leg: the FULLRESYNC RDB carries a moon-graph-store aux blob. The master freezes every graph's write buffer to CSR segments (the checkpoint's own "freeze is the only serialization path" contract) and ships to_bytes() encodings + id-allocation cursors; the replica installs them exactly like restart recovery — authoritative replace, id floors restored (replication::graph_sync, format v1, bounds-checked). An EMPTY blob (master has no graphs) drops replica-local graphs; an ABSENT aux (pre-graph-sync master) warns and keeps. Mmap (restart-loaded) segments export their mapped bytes verbatim (CsrStorage::to_bytes Mmap arm was a silent empty Vec). READONLY guard is now Cypher-aware: read-only GRAPH.QUERY (MATCH/RETURN) is served by replicas; only write queries are rejected (is_cypher_write_query token scan — false-positives route to rejection, never false-negatives). Previously the blanket W flag rejected ALL GRAPH.QUERY on replicas. Applied to monoio, sharded-tokio, and single handlers. New e2e tests/replication_graph.rs: - replica_syncs_graph_live_stream — node/property/GRAPH.LIST parity, plus a zero-reconnect stream-health assertion (a reconnect/resync loop delivers the same data via repeated snapshots and would mask a dead live stream — the exact failure mode fixed in the previous commit). - replica_backfills_graph_from_snapshot — pre-existing graph arrives via the snapshot, post-snapshot writes keep flowing live. Unit tests: graph_sync round-trip (nodes/edges/props fidelity, authoritative replace, empty-store drop, truncated-blob rejection). author: Tin Dang
…iew P0/P1 Fixes every confirmed finding from the adversarial review of the self-SPSC repair (8e6cb74) and graph-plane replication (dc603a3): P0-1 — MULTI/EXEC bodies never replicated at --shards 1. EXEC persisted its body through persist_txn_aof's AOF-only leg, the ONE local write path that skipped the replication plane: the master committed durably and the replica saw nothing (deterministic silent divergence). The txn body now records each entry via record_local_write in the same synchronous stretch as the (fully synchronous) transaction execution, and persist_txn_aof takes a repl_recorded flag (lsn = 0, no offset double-advance — same contract as the single-command legs). New e2e replica_applies_multi_exec_bodies, red before / green after; the double INCR doubles as a double-apply canary. P0-2 — FULLRESYNC snapshot capture raced undrained local writes. The prior design deferred backlog append + offset advance + live fan-out to one queued event-loop message, so a mutation could be INSIDE the RDB capture while still below the advertised snapshot offset — re-delivered via backlog catch-up, double-applying non-idempotent commands. Structural fix: record_local_write appends the backlog bytes and advances the shard offset SYNCHRONOUSLY at write time (mutation and replication ledger are one no-await stretch); only the live replica try_send defers (new ShardMessage::ReplicaLiveFanout, replacing ReplicateVerbatim). RegisterReplica now carries push_offset captured at push time — an offset read at drain could cover a write whose fan-out message is queued behind the registration (catch-up + live = twice). Exactly-once now holds for every write/attach interleave; the send_backlog_range/registration barrier protocol is unchanged. P1-5 (+2 adjacent gaps found while fixing it) — the graph snapshot blob lost all soft state living outside CSR segments: (a) the CSR byte format has no validity section, so segment edge tombstones exist only in the in-memory overlay (BOTH Heap and Mmap variants — the review flagged Mmap, but heap to_bytes drops them identically); (b) freeze() RETAINS cross-tier delta edges in the write buffer, so they are in NO segment; (c) copy-up dead-shadow nodes tombstoning frozen rows never freeze either. The master recovers all three from its WAL on restart; a replica has no WAL — it resurrected deleted edges/nodes and silently lost every cross-tier edge. Blob format v2 ships a per-segment deleted-edge sidecar, the retained delta edges (original edge ids so later streamed REMOVEEDGE records resolve; weight/props via the existing encode_edge_record codec), and the dead-shadow list; install re-applies all three (restore_dead_shadows now pub(crate)). Unit test validity_overlay_delta_edges_and_dead_shadows_survive_export covers all three sections plus the default-weight empty-record path. P1-4 — streamed graph replay was O(N²): every replicated GRAPH.* record re-scanned all write-buffer nodes AND all segment rows to pre-seed the replay node_map (unbounded replication lag on bulk graph loads, backlog eviction, resync storms). The map is gone: node existence resolves lazily via the external_id <-> NodeKey bijection (O(1) write-buf hash probe + per-segment MPH lookup), and the id-allocation floor is raised per segment header max_node_id (new CsrStorage::max_node_id) instead of per row. Restart recovery takes the same path — 541 graph lib tests + 38 freeze-boundary tests green. P1-6 — FULLRESYNC graph export (inside the synchronous with_shard block) no longer freezes untouched write buffers on every reconnect-driven resync; a freeze that fails with live nodes present now warns loudly instead of silently exporting a partial graph. P2-7 — the shard self-queue drains unbounded per cycle: entries are cheap try_sends/vec pushes, and a MAX_DRAIN_PER_CYCLE cut-off could strand a replica's live bytes for a full tick. P1-3 (drain-order inversion) is structurally resolved by the P0-2 fix: backlog bytes are now appended at mutation time, in mutation order, so execute_batch-vs-self-queue drain ordering no longer affects the replication stream. Verification: lib 4110 pass; e2e replication_streaming 4/4 (incl new txn test), replication_graph 2/2, replication_test 5/5, replication_hardening 5/5; graph lib 541 + graph_freeze_boundary 38; fmt + clippy clean on default and tokio feature sets; tokio lib replication tests 43 pass. author: Tin Dang
…lay liveness, blob endpoint guard Adversarial round-2 review of fa193ba confirmed every round-1 fix correct under interleaving analysis, and surfaced four remaining gaps. This commit addresses all of them: Finding B (P0/P1) — TEMPORAL.INVALIDATE never replicated. The handler drained the graph WAL but only fed the local WAL leg; a replica kept valid_to = ∞ for entities the master invalidated. The drained GraphTemporal record is a binary wal_v3 payload the RESP replication link can't carry, and replaying the USER command would re-capture wall_ms on the replica (valid_to divergence). The master now streams a deterministic wall-clock-pinned internal form: TEMPORAL.INVALIDATE-AT <graph> <N|E> <entity_id> <wall_ms> (serialize_invalidate_at / parse_invalidate_at in command/temporal.rs), recorded via record_local_write in the same synchronous stretch as the mutation — identical contract to the GRAPH.* leg. The replica applies it through the same apply_invalidate the master ran, dropping the locally-drained WAL payload (matching apply_graph's no-local-persist model). New e2e REPL-GRAPH-03 (replica_applies_temporal_invalidate) verifies temporal visibility converges via VALID_AT far-future queries — confirmed RED with the master leg reverted, GREEN with it. Finding F (P1) — streamed replay could resurrect a tombstoned node. The P1-4 lazy-resolver rewrite made node_exists accept DEAD write-buf entries (get_node does not filter deleted_lsn), so a stray SETPROP for a node removed in an EARLIER streamed replay call re-registered it into the live property index (set_node_property has no aliveness guard). Split the predicate: - node_present (AddNode dedup): ANY record of the id — live, dead tombstone/shadow, or CSR row. Slotmap ids are never reused, so an AddNode for a present id is only ever a full-history replay. - node_alive (edge endpoints, SETPROP, SETLABEL, REMOVENODE): a write-buffer entry is authoritative (deleted_lsn == MAX ⇒ live); only absent ids fall back to CSR rows. Restores the old pre-seeded map's live-only iter_nodes() semantics and additionally closes the dead-shadow variant of the same hole. New unit test test_streamed_replay_setprop_after_remove_does_not_ resurrect (two sequential replay_into calls) — RED before, GREEN after. Finding E (P2) — graph snapshot install trusted blob delta edges. add_edge_across_tiers_with_id's aliveness check only fires for RESIDENT endpoints and the write buffer is empty at install time, so a corrupted blob referencing a nonexistent node installed silently. The install loop now verifies both endpoints against the just-installed segments and drops the edge with tracing::warn! otherwise. Finding A (P0, known limitation, NOT fixed here) — WS.*/MQ.* writes never reach the replication plane. Full support needs deterministic record forms (WS.CREATE mints a fresh UUIDv7 per execution — verbatim streaming would diverge), replica apply arms, and snapshot coverage — tracked as follow-up task #34 together with the pre-existing Lua-EVAL (finding C) and expiry/eviction (finding D) gaps. Interim: a one-time warn_unreplicated_plane fires when a WS/MQ mutation executes while a replica is attached, so operators learn about the divergence at write time instead of at failover. Finding G (P2, note only): the P0-1 EXEC fix pushes one ReplicaLiveFanout per body entry in a synchronous burst through the unbounded self-queue drain — documented as a tail-latency vector at the call site; each drain iteration is a try_send, so cheap unless EXEC bodies grow unbounded. Verification: lib 4111 green (both after fmt), replication_graph 3/3 (incl. new REPL-GRAPH-03), replication_streaming 4/4, replication_hardening 5/5, replication_test (tokio) 5/5, tokio lib replication 43/43, fmt clean, clippy clean on default and runtime-tokio,jemalloc feature sets. author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
📝 WalkthroughWalkthroughReplication now supports graph live streaming and snapshot installation, deterministic temporal invalidation, tombstone-safe replay, single-shard fanout ordering, MULTI/EXEC replication, Cypher-aware READONLY handling, and warnings for unsupported WS/MQ replication paths. ChangesReplication and graph-plane changes
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Master
participant Replication
participant Replica
participant GraphStore
Client->>Master: GRAPH.* or TEMPORAL.INVALIDATE
Master->>GraphStore: apply mutation and produce WAL
Master->>Replication: record backlog bytes and shard offset
Replication->>Replica: stream live record
Replica->>GraphStore: replay deterministic record
Possibly related PRs
Suggested labels: 🚥 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: 4
🤖 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 143-146: Clarify the CHANGELOG entry for
tests/replication_graph.rs as ignored/manual coverage, explicitly noting it is
not run by standard CI because the test is marked #[ignore]; alternatively, add
an explicit CI job invoking replication_graph with --ignored and document that
coverage.
In `@src/command/temporal.rs`:
- Around line 86-148: Add a unit-test module for serialize_invalidate_at and
parse_invalidate_at covering successful node and edge round trips, including a
negative wall_ms, and asserting malformed argument lists or fields return None.
Use the existing Frame and Bytes types to construct parser inputs and verify all
decoded values match the originals.
In `@src/replication/master.rs`:
- Around line 750-758: Ensure replica registration is cleaned up when catch-up
fails between push_register_replica_inline and drain_replica_inline_single_shard
in both FullResync and PartialResync. Handle send_backlog_range errors by
enqueueing or otherwise performing UnregisterReplica before propagating the
error, while preserving existing cleanup for later drain failures.
In `@tests/replication_graph.rs`:
- Around line 134-135: The replication graph test uses ports shared with
replication_streaming, causing parallel test conflicts. Update the master_addr
and replica_addr values in the replication graph test to a unique,
non-overlapping port pair.
🪄 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: b9135d5f-b5c8-42c9-9087-639c8f240890
📒 Files selected for processing (31)
CHANGELOG.mdsrc/command/temporal.rssrc/graph/csr/mmap.rssrc/graph/csr/storage.rssrc/graph/replay.rssrc/graph/store.rssrc/persistence/redis_rdb.rssrc/replication/apply.rssrc/replication/backlog.rssrc/replication/graph_sync.rssrc/replication/master.rssrc/replication/mod.rssrc/replication/state.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/ft.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_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/conn_accept.rssrc/shard/dispatch.rssrc/shard/event_loop.rssrc/shard/mod.rssrc/shard/self_msg.rssrc/shard/spsc_handler.rstests/replication_graph.rstests/replication_streaming.rs
💤 Files with no reviewable changes (1)
- src/shard/conn_accept.rs
Comment on lines
+143
to
+146
| - New e2e `tests/replication_graph.rs`: live-stream parity (nodes, properties, | ||
| GRAPH.LIST) with a zero-reconnect stream-health assertion that would have | ||
| caught the masked dead stream, plus snapshot backfill + post-snapshot live | ||
| growth. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Clarify this as ignored/manual coverage. tests/replication_graph.rs is #[ignore], and CI only runs replication_hardening, so the zero-reconnect assertion won’t run in standard test jobs. Either note that here or add an explicit --ignored job for replication_graph.
🤖 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 143 - 146, Clarify the CHANGELOG entry for
tests/replication_graph.rs as ignored/manual coverage, explicitly noting it is
not run by standard CI because the test is marked #[ignore]; alternatively, add
an explicit CI job invoking replication_graph with --ignored and document that
coverage.
Comment on lines
+86
to
+148
| /// Serialize the deterministic, wall-clock-pinned replication form of | ||
| /// TEMPORAL.INVALIDATE (v0.7 graph replication, adversarial round-2 finding | ||
| /// B): `TEMPORAL.INVALIDATE-AT <graph> <N|E> <entity_id> <wall_ms>`. | ||
| /// | ||
| /// The user command captures `wall_ms` at execution time, so streaming it | ||
| /// verbatim would let master and replica disagree on `valid_to`; and the | ||
| /// drained `GraphTemporal` WAL record is a binary wal_v3 payload the RESP | ||
| /// replication link cannot carry. This internal RESP form pins the master's | ||
| /// wall clock; the replica applies it via `apply_invalidate` with the SAME | ||
| /// `wall_ms` (see `replication::apply`). | ||
| #[cfg(feature = "graph")] | ||
| pub fn serialize_invalidate_at( | ||
| graph_name: &[u8], | ||
| is_node: bool, | ||
| entity_id: u64, | ||
| wall_ms: i64, | ||
| ) -> Vec<u8> { | ||
| fn write_bulk(buf: &mut Vec<u8>, data: &[u8]) { | ||
| let mut n = itoa::Buffer::new(); | ||
| buf.push(b'$'); | ||
| buf.extend_from_slice(n.format(data.len()).as_bytes()); | ||
| buf.extend_from_slice(b"\r\n"); | ||
| buf.extend_from_slice(data); | ||
| buf.extend_from_slice(b"\r\n"); | ||
| } | ||
| let mut id_buf = itoa::Buffer::new(); | ||
| let mut ms_buf = itoa::Buffer::new(); | ||
| let mut buf = Vec::with_capacity(96 + graph_name.len()); | ||
| buf.extend_from_slice(b"*5\r\n"); | ||
| write_bulk(&mut buf, b"TEMPORAL.INVALIDATE-AT"); | ||
| write_bulk(&mut buf, graph_name); | ||
| write_bulk(&mut buf, if is_node { b"N" } else { b"E" }); | ||
| write_bulk(&mut buf, id_buf.format(entity_id).as_bytes()); | ||
| write_bulk(&mut buf, ms_buf.format(wall_ms).as_bytes()); | ||
| buf | ||
| } | ||
|
|
||
| /// Parse the argument list of a replicated `TEMPORAL.INVALIDATE-AT` record | ||
| /// (inverse of [`serialize_invalidate_at`], minus the command name). | ||
| /// Returns `(graph_name, is_node, entity_id, wall_ms)` or `None` on any | ||
| /// malformed field — the replica warns and skips rather than diverging | ||
| /// silently on garbage. | ||
| #[cfg(feature = "graph")] | ||
| pub fn parse_invalidate_at(args: &[Frame]) -> Option<(Bytes, bool, u64, i64)> { | ||
| if args.len() != 4 { | ||
| return None; | ||
| } | ||
| let bulk = |f: &Frame| -> Option<Bytes> { | ||
| match f { | ||
| Frame::BulkString(b) | Frame::SimpleString(b) => Some(b.clone()), | ||
| _ => None, | ||
| } | ||
| }; | ||
| let graph_name = bulk(&args[0])?; | ||
| let is_node = match bulk(&args[1])?.as_ref() { | ||
| b"N" => true, | ||
| b"E" => false, | ||
| _ => return None, | ||
| }; | ||
| let entity_id: u64 = std::str::from_utf8(&bulk(&args[2])?).ok()?.parse().ok()?; | ||
| let wall_ms: i64 = std::str::from_utf8(&bulk(&args[3])?).ok()?.parse().ok()?; | ||
| Some((graph_name, is_node, entity_id, wall_ms)) | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Look for existing unit tests exercising the invalidate-at codec.
rg -nP 'serialize_invalidate_at|parse_invalidate_at' -g '*.rs' -C2Repository: pilotspace/moon
Length of output: 153
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n## temporal.rs outline\n'
ast-grep outline src/command/temporal.rs --view expanded || true
printf '\n## relevant code around serialize/parse\n'
sed -n '1,220p' src/command/temporal.rs | cat -n
printf '\n## search for invalidate-at and related tests\n'
rg -n --hidden -S 'TEMPORAL\.INVALIDATE-AT|serialize_invalidate_at|parse_invalidate_at|TEMPORAL\.INVALIDATE|consistency-test|consistency test|unit test' src test tests . || true
printf '\n## command module files\n'
git ls-files 'src/command/**' | sed -n '1,200p'Repository: pilotspace/moon
Length of output: 50371
Add a round-trip unit test for TEMPORAL.INVALIDATE-AT (src/command/temporal.rs:97-148)
serialize_invalidate_at/parse_invalidate_at should have a direct round-trip test covering node/edge cases, negative wall_ms, and malformed-input rejection. The end-to-end replication tests don’t exercise this codec in isolation.
🤖 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/command/temporal.rs` around lines 86 - 148, Add a unit-test module for
serialize_invalidate_at and parse_invalidate_at covering successful node and
edge round trips, including a negative wall_ms, and asserting malformed argument
lists or fields return None. Use the existing Frame and Bytes types to construct
parser inputs and verify all decoded values match the originals.
Source: Coding guidelines
Comment on lines
+750
to
+758
| let reg = push_register_replica_inline(&repl_state)?; | ||
| let reg_offset = reg | ||
| .reg_rx | ||
| .recv_async() | ||
| .await | ||
| .map_err(|_| anyhow::anyhow!("event loop dropped registration reply"))?; | ||
| send_backlog_range(&mut stream, &backlog_slot, snapshot_offset, reg_offset).await?; | ||
|
|
||
| drain_replica_inline_single_shard(reg, replica_addr, stream, repl_state, dispatch_tx) | ||
| .await?; | ||
| drain_replica_inline_single_shard(reg, replica_addr, stream, repl_state).await?; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect wal_append_and_fanout and any replica_txs pruning on failed send.
rg -nP -C4 'fn wal_append_and_fanout' src/shard/
rg -nP -C3 'replica_txs\.(retain|remove|swap_remove)' src/Repository: pilotspace/moon
Length of output: 1421
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- src/shard/spsc_handler.rs around UnregisterReplica / ReplicaLiveFanout / wal_append_and_fanout ---'
sed -n '2538,2595p' src/shard/spsc_handler.rs
echo
sed -n '3378,3455p' src/shard/spsc_handler.rs
echo '--- src/replication/master.rs around the cited resync paths ---'
sed -n '730,790p' src/replication/master.rsRepository: pilotspace/moon
Length of output: 9673
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- registration helper / drain helper in src/replication/master.rs ---'
rg -n -C5 'push_register_replica_inline|drain_replica_inline_single_shard|UnregisterReplica|reg_rx|replica_txs' src/replication/master.rs
echo
echo '--- exact definitions around the helper(s) ---'
sed -n '1,260p' src/replication/master.rsRepository: pilotspace/moon
Length of output: 15282
Replica registration needs cleanup on catch-up failures. Both FullResync and PartialResync register the replica before send_backlog_range(), so an error returns before drain_replica_inline_single_shard() can enqueue UnregisterReplica. Since replica_txs only prunes on that message and fanout doesn’t drop dead senders, the stale entry can stick around for the shard lifetime.
🤖 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 750 - 758, Ensure replica
registration is cleaned up when catch-up fails between
push_register_replica_inline and drain_replica_inline_single_shard in both
FullResync and PartialResync. Handle send_backlog_range errors by enqueueing or
otherwise performing UnregisterReplica before propagating the error, while
preserving existing cleanup for later drain failures.
Comment on lines
+134
to
+135
| let master_addr = "127.0.0.1:16730"; | ||
| let replica_addr = "127.0.0.1:16731"; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- replication_graph.rs around cited lines ---\n'
sed -n '120,160p' tests/replication_graph.rs
printf '\n--- replication_streaming.rs port literals ---\n'
rg -n '16730|16731|master_addr|replica_addr' tests/replication_streaming.rs tests/replication_graph.rs
printf '\n--- all occurrences in tests of the two ports ---\n'
rg -n '127\.0\.0\.1:16730|127\.0\.0\.1:16731|16730|16731' tests
printf '\n--- integration test files list ---\n'
git ls-files 'tests/*.rs'Repository: pilotspace/moon
Length of output: 244
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- replication_graph.rs around cited lines ---'
sed -n '120,160p' tests/replication_graph.rs
echo
echo '--- replication_streaming.rs port literals ---'
rg -n '16730|16731|master_addr|replica_addr' tests/replication_streaming.rs tests/replication_graph.rs
echo
echo '--- all occurrences in tests of the two ports ---'
rg -n '127\.0\.0\.1:16730|127\.0\.0\.1:16731|16730|16731' tests
echo
echo '--- integration test files list ---'
git ls-files 'tests/*.rs'Repository: pilotspace/moon
Length of output: 17491
Use a unique port pair here. tests/replication_streaming.rs also binds 127.0.0.1:16730/16731, so these ignored integration binaries can contend for the same ports when cargo runs them in parallel. Pick a non-overlapping pair for this test.
🤖 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_graph.rs` around lines 134 - 135, The replication graph
test uses ports shared with replication_streaming, causing parallel test
conflicts. Update the master_addr and replica_addr values in the replication
graph test to a unique, non-overlapping port pair.
This was referenced Jul 11, 2026
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
v0.7 graph-plane replication: live GRAPH.* mutation streaming + PSYNC full-resync snapshot backfill, hardened through two adversarial review rounds.
What ships
GRAPH.ADDNODE <g> <node_id> …, FNV-hashed u16 label/prop ids —label_to_idis stateless so master and replica resolve identically). Replicas apply through the sameGraphReplayCollectorrestart recovery uses — no id re-allocation, client-cached handles survive.to_byteswrites no validity section), freeze-retained cross-tier delta edges (original ids + weight/props), and copy-up dead-shadow tombstones — installed in dependency order (floors → segments → delta edges → shadows), fully bounds-checked before any mutation.record_local_writeperforms backlog append + shard-offset advance synchronously at write time — atomic w.r.t. the inline PSYNC snapshot capture; only the livetry_senddefers viaReplicaLiveFanouton the self queue.RegisterReplica.push_offsetcaptured at push time. Proven for all write/attach interleavings (round-2 verified: single/concurrent PSYNC, +CONTINUE, unregister ordering).TEMPORAL.INVALIDATE-AT <graph> <N|E> <entity_id> <wall_ms>sovalid_tomatches exactly on both sides.node_present(AddNode dedup) vsnode_alive(mutation targets) — a stray SETPROP for a tombstoned node can no longer resurrect it into the live property index.Scope
Single-shard master (
--shards 1), matching the R0/R0.5 legs — multi-shard masters ride the R2 PSYNC redesign.Verification
runtime-tokio,jemallocSummary by CodeRabbit
New Features
MULTI/EXECwrites.Bug Fixes
Documentation