diff --git a/CHANGELOG.md b/CHANGELOG.md index a5bc007c8..e0235570f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed — replication round-2 hardening: TEMPORAL.INVALIDATE, replay liveness, blob endpoint checks + +- **TEMPORAL.INVALIDATE never replicated** (round-2 finding B): the handler + drained the graph WAL — the same record mechanism GRAPH.\* replication + uses — but only fed the local WAL, never the replication plane; a replica + silently kept `valid_to = ∞` for entities the master had invalidated. The + master now streams a deterministic, wall-clock-pinned internal form + (`TEMPORAL.INVALIDATE-AT `, single-shard + scope like every replication leg) so master and replica agree on the exact + `valid_to`; the replica applies it through the same `apply_invalidate` the + master ran. New e2e REPL-GRAPH-03 proves temporal visibility converges + (red without the master leg, green with it). +- **Streamed replay could resurrect a tombstoned node** (round-2 finding F, + regression from the P1-4 lazy-resolver rewrite): the lazy `node_exists` + accepted DEAD write-buffer 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. Split + into `node_present` (AddNode dedup — any record of the id, matching the + never-reuse slotmap id contract) and `node_alive` (edge endpoints / + SETPROP / SETLABEL / REMOVENODE — write-buffer entry is authoritative, + live-only, matching the old pre-seeded map's `iter_nodes()` semantics). +- **Graph snapshot install now rejects delta edges with unknown endpoints** + (round-2 finding E, defense-in-depth): `add_edge_across_tiers_with_id`'s + aliveness check only fires for resident endpoints — 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 LOUD (`tracing::warn!`) otherwise. +- **WS.\*/MQ.\* writes are NOT replicated in v0.7 — now fail-loud** (round-2 + finding A, known limitation): WS.CREATE/WS.DROP and MQ mutations persist + durably on the master (WAL) but have no deterministic replication record + form yet (WS.CREATE mints a fresh UUIDv7 per execution — verbatim + streaming would diverge). A one-time `tracing::warn!` now fires when such + a write executes while a replica is attached, instead of silent divergence + discovered at failover. Full support (id-pinned record forms + replica + apply arms + snapshot coverage) is tracked as follow-up work, alongside + the pre-existing Lua-EVAL and expiry/eviction propagation gaps. + ### Fixed — CLIENT TRACKING dead on the monoio runtime (H-3 reorder regression) - Since the H-3 ACL reorder (#258), `CLIENT TRACKING ON|OFF` answered @@ -20,6 +57,94 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 post-ACL, so the H-3 deniability guarantee is unchanged). Re-greens all 5 `client_tracking_invalidation` black-box tests on monoio. +### Fixed — replication exactly-once + txn/graph fidelity (adversarial-review P0/P1) + +- **MULTI/EXEC bodies never replicated at `--shards 1`** (P0-1): EXEC persisted + its body through `persist_txn_aof`'s AOF-only leg — the one local write path + that skipped the replication plane. A `MULTI/SET/EXEC` committed durably on + the master and never reached the replica: silent deterministic divergence + for every application using transactions. The txn body now records each + entry through the same `record_local_write` leg as single-command writes + (AOF `lsn = 0`, no double-advance). New e2e + `replica_applies_multi_exec_bodies` (INCR doubles as a double-apply canary). +- **FULLRESYNC snapshot capture raced undrained local writes** (P0-2): the + original design queued backlog append + offset advance + live fan-out as ONE + deferred event-loop message, so a mutation could sit inside the RDB while + still below the advertised snapshot offset — re-delivered via backlog + catch-up and double-applied (INCR/LPUSH divergence). `record_local_write` + now appends the backlog bytes and advances the shard offset SYNCHRONOUSLY + at write time (atomic with the mutation w.r.t. the inline PSYNC capture); + only the live replica `try_send` is deferred (`ReplicaLiveFanout`). + `RegisterReplica` correspondingly carries a push-time offset so catch-up + and live delivery stay disjoint for every write/attach interleave. +- **Graph snapshot lost soft state that lives outside CSR segments** (P1-5 + + two adjacent gaps): the CSR byte format has no validity section, `freeze()` + RETAINS cross-tier delta edges in the write buffer, and copy-up node + tombstones never freeze — the master recovers all three from its WAL on + restart, but a replica has no WAL, so 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), and the + dead-shadow list; install re-applies all three. +- **Streamed graph replay was O(N²)** (P1-4): each replicated GRAPH.* record + re-scanned every write-buffer node and every segment row to pre-seed the + replay id map — unbounded replication lag on bulk graph loads. Node + existence is now resolved lazily (O(1) write-buf probe + MPH segment + lookup), and the id-allocation floor is raised per segment header max + instead of per row. +- Also: FULLRESYNC graph export skips freezing untouched write buffers + (P1-6, repeated-resync latency), and the shard self-queue is drained + unbounded per cycle (P2-7 — entries are cheap try_sends; a cycle cap could + strand a replica's live bytes by a full tick). + +### Fixed — single-shard live replication stream was DEAD (self-SPSC gap) + +- **The R0 live stream never actually flowed at `--shards 1`.** 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: 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. Tests stayed green because each resync's RDB + carried the latest keyspace + FT defs — data crawled across via snapshot + polling, masking the dead stream. Fixed with a thread-local self-message + queue (`shard::self_msg`) drained by the event loop alongside its SPSC + consumers; PSYNC registration, FT.*/graph fan-out, and local-write fan-out + all route through it. +- **Local (same-shard) writes now feed the replication plane.** Successful + local writes push their wire bytes as `ReplicateVerbatim` before any await + (mutation + record are one synchronous stretch, atomic w.r.t. snapshot + capture); the drained message does backlog + offset + replica fan-out + together, and the AOF leg no longer double-advances the offset (lsn = 0 when + fan-out owns the advance). +- **Replication backlog now seeds at the current shard offset** on lazy + allocation (`ReplicationBacklog::new_at`) — an unseeded backlog made every + catch-up range read on a pre-written master fail as "evicted". +- New process-global `fanout_hint_active()` (one Relaxed load, set on first + replica attach, never cleared) gates all fan-out serialization so + non-replicating servers pay nothing on the hot path. + +### Added — v0.7 graph-plane replication (live stream + snapshot backfill) + +- **Live leg:** graph mutations (GRAPH.* + Cypher writes) stream to replicas + as their deterministic, id-pinned WAL records (`GRAPH.ADDNODE …`; + label/prop ids are a stateless FNV hash, identical on both sides). The + replica applies them through the same `GraphReplayCollector` restart + recovery uses — no id re-allocation, no divergence. Replay's edge/SET + resolution now also seeds from write-buffer-resident nodes so one-record-at- + a-time streaming replay resolves endpoints applied by earlier records. +- **Snapshot leg:** the FULLRESYNC RDB carries a `moon-graph-store` aux blob — + every graph's write buffer is frozen to CSR segments (the checkpoint's own + "freeze is the only serialization path" contract) and shipped as + `to_bytes()` encodings + id cursors; the replica installs them exactly like + restart recovery (`replication::graph_sync`). Mmap (restart-loaded) segments + export their mapped bytes verbatim. +- **READONLY guard is now Cypher-aware:** a read-only `GRAPH.QUERY` + (MATCH/RETURN) is served by replicas; only write queries (CREATE/DELETE/ + SET/MERGE tokens) are rejected. Previously the blanket `W` flag rejected all + GRAPH.QUERY on replicas. +- 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. + ### Fixed — PSYNC attach races closed (adversarial-review findings on R0/R0.5) - **Registration-bounded catch-up:** the master now registers the replica with diff --git a/src/command/temporal.rs b/src/command/temporal.rs index 22ed45cfe..71302c723 100644 --- a/src/command/temporal.rs +++ b/src/command/temporal.rs @@ -83,6 +83,70 @@ pub fn validate_invalidate(args: &[Frame]) -> Result<(u64, bool, Bytes), Frame> Ok((entity_id, is_node, graph_name)) } +/// Serialize the deterministic, wall-clock-pinned replication form of +/// TEMPORAL.INVALIDATE (v0.7 graph replication, adversarial round-2 finding +/// B): `TEMPORAL.INVALIDATE-AT `. +/// +/// 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 { + fn write_bulk(buf: &mut Vec, 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 { + 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)) +} + /// Apply a TEMPORAL.INVALIDATE mutation to a graph store. /// /// Sets `valid_to = wall_ms` on the entity and pushes the WAL payload into diff --git a/src/graph/csr/mmap.rs b/src/graph/csr/mmap.rs index d5e722260..a798340d0 100644 --- a/src/graph/csr/mmap.rs +++ b/src/graph/csr/mmap.rs @@ -385,6 +385,14 @@ impl MmapCsrSegment { }) } + /// The complete serialized segment bytes (the mapped file IS the + /// `to_bytes()` encoding — `write_to_file` produced it). Used by the + /// replication snapshot export so a restart-loaded (mmap) segment can be + /// shipped to a replica with full fidelity. + pub fn raw_bytes(&self) -> &[u8] { + &self._mmap + } + /// Node property blob (borrowed from mmap; empty for pre-v5 files). pub fn node_props_blob(&self) -> &[u8] { if self.node_props_len == 0 { diff --git a/src/graph/csr/storage.rs b/src/graph/csr/storage.rs index 748bbef91..bdefec929 100644 --- a/src/graph/csr/storage.rs +++ b/src/graph/csr/storage.rs @@ -211,6 +211,17 @@ impl CsrStorage { } } + /// Largest frozen external node id (header field). O(1) — lets recovery + /// and streamed replay raise the id-allocation floor per SEGMENT instead + /// of per node (`ensure_node_id_floor` keeps a single monotonic counter, + /// so the max subsumes every row's id). + pub fn max_node_id(&self) -> u64 { + match self { + CsrStorage::Heap(s) => s.header.max_node_id, + CsrStorage::Mmap(s) => s.header.max_node_id, + } + } + /// Access the validity bitmap. pub fn validity(&self) -> &RoaringBitmap { match self { @@ -521,10 +532,20 @@ impl CsrStorage { } /// Serialize to bytes (only meaningful for Heap variant). + /// + /// ⚠ Post-load `mark_deleted` tombstones live in the in-memory validity + /// overlay and are NOT captured for EITHER variant — the byte format has + /// no validity section (`to_bytes` writes `validity_bitmap_offset = 0`). + /// Consumers that need deletion fidelity across the wire must ship the + /// overlay separately (`replication::graph_sync` writes a deleted-edge + /// sidecar per segment and re-applies it on install). pub fn to_bytes(&self) -> Vec { match self { CsrStorage::Heap(s) => s.to_bytes(), - CsrStorage::Mmap(_) => Vec::new(), // Not applicable + // The mapped file is byte-identical to the `to_bytes()` encoding + // (`write_to_file` produced it), so a copy of the mapped region + // round-trips through `CsrSegment::from_bytes`. + CsrStorage::Mmap(s) => s.raw_bytes().to_vec(), } } diff --git a/src/graph/replay.rs b/src/graph/replay.rs index 6a3a2b46c..914f02797 100644 --- a/src/graph/replay.rs +++ b/src/graph/replay.rs @@ -501,20 +501,54 @@ impl GraphReplayCollector { }; let (mut mg, immutable) = take_memgraph(graph); - // Seed node_maps from immutable CSR segments so edges referencing - // CSR-resident nodes (loaded during recovery) can be resolved. - // Also raise the id-allocation floor past every frozen - // external_id so fresh post-replay inserts can never alias - // a frozen row. - let mut node_map: HashMap = HashMap::new(); + // Raise the id-allocation floor past every frozen external_id + // so fresh post-replay inserts can never alias a frozen row. + // Per SEGMENT (header max), not per node: the floor is one + // monotonic counter, so the max subsumes every row's id — and + // a replica replays streamed WAL records ONE AT A TIME + // through this function, so a per-node pass here would cost + // O(total nodes) per record (adversarial-review P1-4). for csr_seg in &immutable { - for nm in csr_seg.node_meta() { - let key_data = slotmap::KeyData::from_ffi(nm.external_id); - let node_key = crate::graph::types::NodeKey::from(key_data); - node_map.insert(nm.external_id, node_key); - mg.ensure_node_id_floor(nm.external_id); - } + mg.ensure_node_id_floor(csr_seg.max_node_id()); } + // Node existence is resolved LAZILY, not via a pre-seeded + // map: external_id ↔ NodeKey is the same KeyData bijection + // used at insert (`add_node_with_id`), so probing the write + // buffer (O(1) hash — covers nodes landed by EARLIER streamed + // replay calls AND this batch's inserts) and the CSR segments + // (MPH lookup) IS the membership test the old map answered. + // The pre-seeded map re-scanned every write-buf node and + // every segment row on each call — O(N²) for an N-record + // stream. + let nk_of = |id: u64| -> crate::graph::types::NodeKey { + crate::graph::types::NodeKey::from(slotmap::KeyData::from_ffi(id)) + }; + // Presence: ANY record of this id — live write-buf node, dead + // tombstone/shadow, or frozen CSR row. AddNode dedup keys off + // presence: slotmap ids are never reused (version bits bump on + // slot reuse), so an AddNode for a present id can only be a + // full-history replay of the same node — re-inserting would + // shadow the existing identity. + let node_present = |mg: &MemGraph, nk: crate::graph::types::NodeKey| -> bool { + mg.get_node(nk).is_some() + || immutable.iter().any(|s| s.lookup_node(nk).is_some()) + }; + // Liveness: mutation targets (edge endpoints, SETPROP/SETLABEL, + // REMOVENODE) must be ALIVE. A write-buffer entry is + // authoritative — `deleted_lsn == MAX` ⇒ live; a tombstoned + // write-buf-only node or a copy-up dead shadow of a frozen row + // ⇒ dead (matching the old live-only `iter_nodes()` map — + // adversarial round-2 finding F: `get_node` alone returns dead + // entries too, and `set_node_property` has no aliveness guard, + // so a stray SETPROP would re-register a tombstoned node into + // the live property index). Only ids with NO write-buf entry + // fall back to the CSR rows. + let node_alive = |mg: &MemGraph, nk: crate::graph::types::NodeKey| -> bool { + match mg.get_node(nk) { + Some(n) => n.deleted_lsn == u64::MAX, + None => immutable.iter().any(|s| s.lookup_node(nk).is_some()), + } + }; // Insert nodes. // Precompute _key property ID for graph expansion mapping. @@ -539,18 +573,17 @@ impl GraphReplayCollector { // the frozen row with a redundant mutable copy (P0: // restart NodeKey aliasing). Keep the CSR-seeded // identity and skip the insert. - let nk = if let Some(&existing) = node_map.get(node_id) { - existing + let candidate = nk_of(*node_id); + let nk = if node_present(&mg, candidate) { + candidate } else { - let nk = mg.add_node_with_id( + mg.add_node_with_id( *node_id, labels.clone(), properties.clone(), embedding.clone(), 0, - ); - node_map.insert(*node_id, nk); - nk + ) }; // Track _key properties for registration after memgraph is // returned. Re-derived even for dedup-skipped (CSR-resident) @@ -581,14 +614,14 @@ impl GraphReplayCollector { .. } = &self.commands[idx] { - let src_key = node_map.get(src_id).copied(); - let dst_key = node_map.get(dst_id).copied(); - if let (Some(src), Some(dst)) = (src_key, dst_key) { - // Cross-tier aware: endpoints seeded from CSR + let src = nk_of(*src_id); + let dst = nk_of(*dst_id); + if node_alive(&mg, src) && node_alive(&mg, dst) { + // Cross-tier aware: endpoints frozen into CSR // segments are non-resident in `mg` — a plain // add_edge would silently drop the edge on - // replay. node_map membership IS the existence - // proof (replayed node or CSR row). The ORIGINAL + // replay. `node_exists` (write buffer OR CSR row) + // is the existence proof. The ORIGINAL // logged edge id is preserved so client-cached // edge handles (and later REMOVEEDGE records) // resolve after restart. @@ -630,13 +663,14 @@ impl GraphReplayCollector { value, .. } => { - let Some(nk) = node_map.get(entity_id).copied() else { + let nk = nk_of(*entity_id); + if !node_alive(&mg, nk) { tracing::warn!( - "WAL replay: SETPROP node_id={} not found in WAL or CSR", + "WAL replay: SETPROP node_id={} not found live in WAL or CSR", entity_id ); continue; - }; + } // CSR-resident target (frozen before the SET was // logged): copy the row up first, like live W2-2. if mg.get_node(nk).is_none() @@ -681,13 +715,14 @@ impl GraphReplayCollector { } } GraphCommand::SetLabel { node_id, label, .. } => { - let Some(nk) = node_map.get(node_id).copied() else { + let nk = nk_of(*node_id); + if !node_alive(&mg, nk) { tracing::warn!( - "WAL replay: SETLABEL node_id={} not found in WAL or CSR", + "WAL replay: SETLABEL node_id={} not found live in WAL or CSR", node_id ); continue; - }; + } if mg.get_node(nk).is_none() && !crate::graph::store::copy_up_into(&mut mg, &immutable, nk) { @@ -707,7 +742,8 @@ impl GraphReplayCollector { // Remove nodes. for &idx in &epoch.remove_node_indices { if let GraphCommand::RemoveNode { node_id, .. } = &self.commands[idx] { - if let Some(nk) = node_map.get(node_id).copied() { + let nk = nk_of(*node_id); + if node_alive(&mg, nk) { if mg.remove_node(nk, 0) { *replayed += 1; } else if mg.get_node(nk).is_none() { @@ -1172,6 +1208,49 @@ mod tests { assert!(node.labels.contains(&5), "SETLABEL applied"); } + /// Adversarial round-2 finding F: a write-buffer-only node removed in an + /// EARLIER replay call (a streamed replica applies ONE record per call) + /// must NOT be resurrected by a later SETPROP for the same id — + /// `set_node_property` has no aliveness guard and would re-register the + /// tombstoned node into the live property index. + #[test] + fn test_streamed_replay_setprop_after_remove_does_not_resurrect() { + let n1 = (1u64 << 32) | 7; + let n1s = n1.to_string(); + let mut store = GraphStore::new(); + + // Call 1 (streamed batch): create + addnode + removenode. + let mut c1 = GraphReplayCollector::new(); + assert!(c1.collect_command(b"GRAPH.CREATE", &[b"g"])); + assert!(c1.collect_command(b"GRAPH.ADDNODE", &[b"g", n1s.as_bytes(), b"1", b"3", b"0"])); + assert!(c1.collect_command(b"GRAPH.REMOVENODE", &[b"g", n1s.as_bytes()])); + assert_eq!(c1.replay_into(&mut store), 3); + + // Call 2 (a LATER streamed record): stray SETPROP for the dead id. + let mut c2 = GraphReplayCollector::new(); + assert!(c2.collect_command( + b"GRAPH.SETPROP", + &[b"g", b"N", n1s.as_bytes(), b"9", b"i", b"1"] + )); + let replayed = c2.replay_into(&mut store); + assert_eq!( + replayed, 0, + "SETPROP on a dead node must skip, not resurrect" + ); + + let g = store.get_graph(b"g").expect("graph"); + let nk = crate::graph::types::NodeKey::from(slotmap::KeyData::from_ffi(n1)); + let node = g + .write_buf + .get_node(nk) + .expect("temporal tombstone entry is retained"); + assert_ne!(node.deleted_lsn, u64::MAX, "node stays dead"); + assert!( + node.properties.iter().all(|(k, _)| *k != 9), + "no property applied to a dead node" + ); + } + /// W2-9 + W2-2: a SETPROP whose target froze before the crash must copy /// the row up into write_buf (frozen rows are immutable) and mutate the /// shadow. diff --git a/src/graph/store.rs b/src/graph/store.rs index 8cc51524a..9b06ff771 100644 --- a/src/graph/store.rs +++ b/src/graph/store.rs @@ -156,7 +156,15 @@ impl NamedGraph { /// Re-materialize copy-up tombstones after a freeze drained them: /// copy the frozen row's data back into the write buffer, then re-delete /// it at its ORIGINAL deletion LSN (preserves MVCC time-travel). - fn restore_dead_shadows(&mut self, dead_shadows: &[(crate::graph::types::NodeKey, u64)]) { + /// + /// `pub(crate)` for the replication snapshot install + /// (`replication::graph_sync`): a replica receiving a graph blob must + /// re-materialize the master's copy-up tombstones the same way, or the + /// frozen rows they shadow resurrect on the replica. + pub(crate) fn restore_dead_shadows( + &mut self, + dead_shadows: &[(crate::graph::types::NodeKey, u64)], + ) { if dead_shadows.is_empty() { return; } diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index 08b50a087..5003fbfd7 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -445,6 +445,10 @@ pub const MOON_AUX_VECTOR_DEFS: &[u8] = b"moon-vector-defs"; /// (`text::index_persist::serialize_text_index_metas` bytes). See /// [`MOON_AUX_VECTOR_DEFS`]. pub const MOON_AUX_TEXT_DEFS: &[u8] = b"moon-text-defs"; +/// Moon replication aux key: whole graph-store snapshot +/// (`replication::graph_sync::export_graph_store` blob — frozen CSR segment +/// encodings + id cursors per graph). See [`MOON_AUX_VECTOR_DEFS`]. +pub const MOON_AUX_GRAPH_STORE: &[u8] = b"moon-graph-store"; /// `write_rdb_refs` plus moon-private AUX fields written immediately after /// the standard header aux block (before any SELECTDB), which is what lets diff --git a/src/replication/apply.rs b/src/replication/apply.rs index 5cb48a76f..5277d16d8 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -184,6 +184,30 @@ pub(crate) fn apply_local(rc: &ReplCommand) -> bool { return; } + // GRAPH.* mutations (v0.7 graph replication) arrive as the master's + // DETERMINISTIC WAL-record form: id-pinned (GRAPH.ADDNODE + // …) with FNV-hashed u16 label/prop-key ids — `label_to_id` + // is a stateless hash, so both sides resolve the same strings to the + // same ids. Generic `dispatch()` parses the USER syntax and would + // re-allocate ids; route through the WAL replay collector instead, + // exactly like restart recovery does. + #[cfg(feature = "graph")] + if crate::graph::replay::GraphReplayCollector::is_graph_command(cmd) { + apply_graph(s, cmd, args); + return; + } + + // TEMPORAL.INVALIDATE arrives as the master's deterministic + // wall-clock-pinned form (`TEMPORAL.INVALIDATE-AT + // `, round-2 finding B) — apply with the SAME + // wall_ms the master used so `valid_to` matches exactly. Generic + // `dispatch()` does not know this internal record. + #[cfg(feature = "graph")] + if cmd.eq_ignore_ascii_case(b"TEMPORAL.INVALIDATE-AT") { + apply_temporal_invalidate(s, cmd, args); + return; + } + // MOVE / cross-db COPY touch two databases at once and are intercepted // BEFORE generic dispatch on the master (see `spsc_two_db`). Generic // `dispatch()` cannot apply them — it returns an error for MOVE and @@ -258,6 +282,81 @@ fn apply_ft( } } +/// Apply one replicated graph WAL record (v0.7 graph replication) into this +/// shard's `GraphStore` through the same `GraphReplayCollector` restart +/// recovery uses. Records arrive one at a time in stream order; the collector +/// resolves edge/SET targets against nodes applied by EARLIER records via the +/// write-buffer seeding in `replay_epoch_aware`. +#[cfg(feature = "graph")] +fn apply_graph(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame]) { + use crate::graph::replay::GraphReplayCollector; + let mut arg_bytes: Vec<&[u8]> = Vec::with_capacity(args.len()); + for a in args { + match a { + Frame::BulkString(b) | Frame::SimpleString(b) => arg_bytes.push(b.as_ref()), + _ => { + tracing::warn!( + "replica apply: non-bulk arg in graph record {} — skipped (graph diverges; \ + full resync required)", + String::from_utf8_lossy(cmd) + ); + return; + } + } + } + let mut collector = GraphReplayCollector::new(); + if !collector.collect_command(cmd, &arg_bytes) { + tracing::warn!( + "replica apply: unparseable graph record {} — skipped (graph diverges; \ + full resync required)", + String::from_utf8_lossy(cmd) + ); + return; + } + if collector.replay_into(&mut s.graph_store) == 0 { + // Not always divergence (e.g. GRAPH.CREATE of an existing graph + // counts 0), but worth surfacing at debug for stream forensics. + tracing::debug!( + "replica apply: graph record {} replayed 0 mutations", + String::from_utf8_lossy(cmd) + ); + } +} + +/// Apply a replicated `TEMPORAL.INVALIDATE-AT` record: same mutation the +/// master ran (`apply_invalidate`) with the master's pinned `wall_ms`. The +/// drained `GraphTemporal` WAL payload is dropped, matching `apply_graph`'s +/// no-local-persistence model (a restarted replica resyncs from the master; +/// leaving it in `wal_pending` would leak into an unrelated later drain). +#[cfg(feature = "graph")] +fn apply_temporal_invalidate(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame]) { + let Some((graph_name, is_node, entity_id, wall_ms)) = + crate::command::temporal::parse_invalidate_at(args) + else { + tracing::warn!( + "replica apply: malformed {} record — skipped (graph diverges; \ + full resync required)", + String::from_utf8_lossy(cmd) + ); + return; + }; + if let Err(e) = crate::command::temporal::apply_invalidate( + &mut s.graph_store, + entity_id, + is_node, + &graph_name, + wall_ms, + ) { + tracing::warn!( + "replica apply: TEMPORAL.INVALIDATE-AT entity_id={} failed: {} \ + (graph diverges; full resync required)", + entity_id, + String::from_utf8_lossy(e) + ); + } + let _ = s.graph_store.drain_wal(); +} + /// Mirror of the master's connection-layer index-parity block /// (`handler_monoio/mod.rs`, "HSET auto-index" onwards): HSET feeds the /// auto-indexer, DEL/UNLINK tombstone, HDEL of a vector field tombstones, @@ -361,12 +460,45 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { // header) carry the FT index DEFINITIONS; standard RDB loaders skip them. let vec_defs = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_VECTOR_DEFS); let text_defs = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_TEXT_DEFS); + #[cfg(feature = "graph")] + let graph_blob = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_GRAPH_STORE); match crate::shard::slice::try_with_shard(|s| { for db in s.databases.iter_mut() { db.clear(); } let loaded = redis_rdb::load_rdb(&mut s.databases, rdb)?; install_snapshot_index_defs(s, vec_defs.as_deref(), text_defs.as_deref()); + // v0.7 graph replication: install the master's whole graph store + // (authoritative replace — an EMPTY blob drops replica-local graphs; + // an ABSENT aux means a pre-graph-sync master, warn-and-keep). + #[cfg(feature = "graph")] + match graph_blob.as_deref() { + Some(blob) => { + match crate::replication::graph_sync::install_graph_store(&mut s.graph_store, blob) + { + Some(n) => { + if n > 0 { + tracing::info!("replica snapshot: installed {} graph(s)", n); + } + } + None => { + return Err(anyhow::anyhow!( + "replica snapshot: malformed graph-store aux blob" + )); + } + } + } + None => { + if s.graph_store.graph_count() > 0 { + tracing::warn!( + "replica snapshot carried no graph-store aux but {} local graph(s) \ + exist — master predates graph replication; keeping local graphs \ + (they may diverge)", + s.graph_store.graph_count() + ); + } + } + } Ok(loaded) }) { Some(r) => r, diff --git a/src/replication/backlog.rs b/src/replication/backlog.rs index 8a3ff678e..250a8f486 100644 --- a/src/replication/backlog.rs +++ b/src/replication/backlog.rs @@ -22,11 +22,23 @@ pub struct ReplicationBacklog { impl ReplicationBacklog { pub fn new(capacity: usize) -> Self { + Self::new_at(capacity, 0) + } + + /// Allocate a backlog whose byte positions begin at `offset`. + /// + /// The backlog is LAZILY allocated on the first replica attach, but the + /// shard offset counter may already be far past zero (local writes advance + /// it via `issue_lsn` even with no replica). Seeding start/end to the + /// current shard offset keeps `bytes_from`/`contains_offset` range math + /// aligned with the counter — an unseeded backlog made every catch-up + /// read on a pre-written master fail as "evicted". + pub fn new_at(capacity: usize, offset: u64) -> Self { ReplicationBacklog { buf: std::collections::VecDeque::with_capacity(capacity), capacity, - start_offset: 0, - end_offset: 0, + start_offset: offset, + end_offset: offset, } } diff --git a/src/replication/graph_sync.rs b/src/replication/graph_sync.rs new file mode 100644 index 000000000..2ad855211 --- /dev/null +++ b/src/replication/graph_sync.rs @@ -0,0 +1,574 @@ +//! v0.7 graph-plane PSYNC snapshot: export / install a shard's `GraphStore` +//! as a moon-private RDB aux blob. +//! +//! The master freezes every graph's write buffer into an immutable CSR +//! segment (the same "freeze is the only serialization path" contract the +//! WAL-v3 checkpoint uses — see `persist_graph_at_checkpoint`) and ships the +//! segments' `to_bytes()` encodings. The replica recreates each graph and +//! injects the segments exactly like restart recovery does +//! (`recover_graph_store`), so node/edge ids, labels, properties, and +//! embeddings all survive with restart-equivalent fidelity. Post-snapshot +//! mutations arrive as deterministic GRAPH.* WAL records on the live stream. +//! +//! Blob format (version 2, little-endian): +//! ```text +//! [u8 version = 2] +//! [u32 graph_count] +//! per graph: +//! [u16 name_len][name bytes] +//! [u64 next_node_id][u64 next_edge_id] (id-allocation cursors) +//! [u32 segment_count] +//! per segment: +//! [u64 blob_len][CsrSegment::to_bytes payload] +//! [u32 deleted_edge_count][u32 edge_idx]* (validity-overlay sidecar) +//! [u32 delta_edge_count] (cross-tier write-buf edges) +//! per delta edge: +//! [u64 edge_ffi][u64 src_ffi][u64 dst_ffi][u16 edge_type][u64 created_lsn] +//! [u32 stored_off][u32 rec_len][encode_edge_record payload] +//! [u32 dead_shadow_count] (copy-up node tombstones) +//! per shadow: [u64 node_ffi][u64 deleted_lsn] +//! ``` +//! +//! The deleted-edge sidecar exists because the CSR byte format does NOT +//! serialize the validity bitmap (`to_bytes` writes `validity_bitmap_offset = +//! 0`; a load initializes all edges valid): soft-deleted edges live only in +//! the in-memory overlay until a physical compaction rewrites the segment. +//! That is fine self-referentially on the master (restart re-applies the +//! deletes from its WAL), but a replica has neither the overlay nor those WAL +//! records — without the sidecar, every edge soft-deleted on the master since +//! the segment was built would be RESURRECTED on the replica and served as +//! live from read queries (adversarial-review P1-5, both `Heap` and `Mmap` +//! variants). + +use bytes::Bytes; + +use crate::graph::csr::{CsrSegment, CsrStorage}; +use crate::graph::segment::GraphSegmentList; +use crate::graph::store::GraphStore; + +const FORMAT_VERSION: u8 = 2; + +/// Export the whole store as a snapshot blob. Freezes every graph's write +/// buffer first (same operation the checkpoint performs; runs on the shard +/// thread between mutations, so the cut is consistent). Always returns a +/// blob — an empty store encodes as `graph_count = 0`, which lets the +/// replica distinguish "master has no graphs" (authoritative: drop local +/// graphs) from "pre-graph-sync master" (aux absent entirely). +pub fn export_graph_store(store: &mut GraphStore) -> Vec { + // Freeze all write buffers so the segments cover the mutable tier. + // Skip graphs whose write buffer is empty: this export runs INSIDE the + // PSYNC handler's synchronous `with_shard` block, blocking the shard's + // event loop — and reconnect-driven full resyncs can arrive repeatedly. + // An unconditional freeze per resync would re-run the dead-shadow scan + + // freeze/thaw machinery over every graph each time even when nothing + // changed (adversarial-review P1-6). + let names: Vec = store.list_graphs().into_iter().cloned().collect(); + for name in &names { + let needs_freeze = store + .get_graph(name) + .is_some_and(|g| g.write_buf.node_count() > 0 || g.write_buf.edge_count() > 0); + if !needs_freeze { + continue; + } + let lsn = store.allocate_lsn(); + if let Some(graph) = store.get_graph_mut(name) { + graph.freeze_and_compact(lsn); + } + } + + let mut buf: Vec = Vec::with_capacity(64); + buf.push(FORMAT_VERSION); + buf.extend_from_slice(&(names.len() as u32).to_le_bytes()); + for name in &names { + let Some(graph) = store.get_graph(name) else { + // Unreachable (names collected above); keep counts honest anyway. + buf.extend_from_slice(&[0u8; 2]); + buf.extend_from_slice(&0u64.to_le_bytes()); + buf.extend_from_slice(&0u64.to_le_bytes()); + buf.extend_from_slice(&0u32.to_le_bytes()); + buf.extend_from_slice(&0u32.to_le_bytes()); + buf.extend_from_slice(&0u32.to_le_bytes()); + continue; + }; + if graph.write_buf.node_count() > 0 { + // A successful freeze drains every live node; nodes still here + // mean the freeze failed (CSR build error) and their data will + // NOT be in the blob. Loud, not silent. + tracing::warn!( + graph = %String::from_utf8_lossy(name), + live_nodes = graph.write_buf.node_count(), + "graph snapshot export: write-buffer freeze failed — \ + unfrozen nodes are missing from the replica snapshot" + ); + } + buf.extend_from_slice(&(name.len() as u16).to_le_bytes()); + buf.extend_from_slice(name); + let (next_node, next_edge) = graph.write_buf.id_cursors(); + buf.extend_from_slice(&next_node.to_le_bytes()); + buf.extend_from_slice(&next_edge.to_le_bytes()); + let segments = graph.segments.load(); + buf.extend_from_slice(&(segments.immutable.len() as u32).to_le_bytes()); + for seg in &segments.immutable { + let blob = seg.to_bytes(); + buf.extend_from_slice(&(blob.len() as u64).to_le_bytes()); + buf.extend_from_slice(&blob); + // Validity-overlay sidecar: `to_bytes` does not serialize the + // validity bitmap, so soft-deleted edges exist only in memory — + // ship the deleted set explicitly (see module doc, P1-5). + let deleted: Vec = (0..seg.edge_count()) + .filter(|&i| !seg.is_valid(i)) + .collect(); + buf.extend_from_slice(&(deleted.len() as u32).to_le_bytes()); + for idx in deleted { + buf.extend_from_slice(&idx.to_le_bytes()); + } + } + // Cross-tier delta edges: freeze() RETAINS edges with a frozen + // endpoint in the write buffer (CsrSegment::from_frozen can't encode + // them), so they are in NO segment — without this section every + // cross-tier edge would silently vanish from the replica. The master + // recovers them from its WAL on restart; the replica has no WAL. + { + use slotmap::Key; + let delta: Vec<_> = graph.write_buf.iter_edges().collect(); + buf.extend_from_slice(&(delta.len() as u32).to_le_bytes()); + for (ek, e) in delta { + buf.extend_from_slice(&ek.data().as_ffi().to_le_bytes()); + buf.extend_from_slice(&e.src.data().as_ffi().to_le_bytes()); + buf.extend_from_slice(&e.dst.data().as_ffi().to_le_bytes()); + buf.extend_from_slice(&e.edge_type.to_le_bytes()); + buf.extend_from_slice(&e.created_lsn.to_le_bytes()); + let mut rec: Vec = Vec::new(); + let stored_off = crate::graph::csr::props::encode_edge_record( + &mut rec, + e.weight, + e.properties.as_ref(), + ); + buf.extend_from_slice(&stored_off.to_le_bytes()); + buf.extend_from_slice(&(rec.len() as u32).to_le_bytes()); + buf.extend_from_slice(&rec); + } + } + // Copy-up tombstones: dead write-buf nodes shadowing a frozen CSR + // row (a node DELETE against a frozen node). Same WAL-vs-no-WAL + // asymmetry as delta edges — without this the deleted node + // RESURRECTS on the replica. Same filter as `freeze_and_compact`'s + // dead-shadow collection. + { + use slotmap::Key; + let shadows: Vec<(u64, u64)> = graph + .write_buf + .iter_dead_nodes() + .filter(|(k, _)| { + segments + .immutable + .iter() + .any(|s| s.lookup_node(*k).is_some()) + }) + .map(|(k, n)| (k.data().as_ffi(), n.deleted_lsn)) + .collect(); + buf.extend_from_slice(&(shadows.len() as u32).to_le_bytes()); + for (ffi, lsn) in shadows { + buf.extend_from_slice(&ffi.to_le_bytes()); + buf.extend_from_slice(&lsn.to_le_bytes()); + } + } + } + buf +} + +/// Install a snapshot blob into `store`, replacing ALL local graph state +/// (authoritative, mirroring the keyspace-replace semantics of RDB load). +/// Returns the number of graphs installed, or `None` on a malformed blob +/// (store is left in whatever partial state was reached — the caller aborts +/// the sync and the replica retries with a fresh full resync). +pub fn install_graph_store(store: &mut GraphStore, blob: &[u8]) -> Option { + let mut cur = Cursor { data: blob, pos: 0 }; + if cur.u8()? != FORMAT_VERSION { + return None; + } + // Authoritative replace: drop everything local first. + let local: Vec = store.list_graphs().into_iter().cloned().collect(); + for name in local { + let _ = store.drop_graph(&name); + } + + let graph_count = cur.u32()? as usize; + for _ in 0..graph_count { + let name_len = cur.u16()? as usize; + let name = Bytes::copy_from_slice(cur.take(name_len)?); + let next_node = cur.u64()?; + let next_edge = cur.u64()?; + let seg_count = cur.u32()? as usize; + + let mut segments: Vec> = Vec::with_capacity(seg_count); + for _ in 0..seg_count { + let len = cur.u64()? as usize; + let bytes = cur.take(len)?; + let mut seg = match CsrSegment::from_bytes(bytes) { + Ok(seg) => seg, + Err(e) => { + tracing::warn!( + graph = %String::from_utf8_lossy(&name), + error = ?e, + "replica graph sync: corrupt CSR segment in snapshot — aborting install" + ); + return None; + } + }; + // Re-apply the master's validity overlay (soft-deleted edges are + // not part of the CSR byte format — see module doc). + let deleted = cur.u32()?; + for _ in 0..deleted { + let idx = cur.u32()?; + if idx >= seg.edge_count() { + return None; // malformed sidecar + } + seg.mark_deleted(idx); + } + segments.push(std::sync::Arc::new(CsrStorage::Heap(seg))); + } + + // Cross-tier delta edges + copy-up tombstones (parsed before the + // graph mutations below so a truncated blob aborts cleanly first). + let delta_count = cur.u32()? as usize; + let mut delta_edges = Vec::with_capacity(delta_count); + for _ in 0..delta_count { + let ek = cur.u64()?; + let src = cur.u64()?; + let dst = cur.u64()?; + let edge_type = cur.u16()?; + let created_lsn = cur.u64()?; + let stored_off = cur.u32()?; + let rec_len = cur.u32()? as usize; + let rec = cur.take(rec_len)?; + let weight = crate::graph::csr::props::decode_edge_weight(rec, stored_off); + let props = crate::graph::csr::props::decode_edge_props(rec, stored_off); + let props = if props.is_empty() { None } else { Some(props) }; + delta_edges.push((ek, src, dst, edge_type, created_lsn, weight, props)); + } + let shadow_count = cur.u32()? as usize; + let mut shadows: Vec<(crate::graph::types::NodeKey, u64)> = + Vec::with_capacity(shadow_count); + for _ in 0..shadow_count { + let ffi = cur.u64()?; + let lsn = cur.u64()?; + shadows.push((slotmap::KeyData::from_ffi(ffi).into(), lsn)); + } + + if store.create_graph(name.clone(), 64_000, 0).is_err() { + return None; + } + let graph = store.get_graph_mut(&name)?; + // Same floor-restore contract as `recover_graph_store`: manifest + // cursors are authoritative, and every frozen external_id raises the + // floor so later inserts can never alias a frozen row. + graph.write_buf.restore_id_cursors(next_node, next_edge); + for seg in &segments { + for nm in seg.node_meta() { + graph.write_buf.ensure_node_id_floor(nm.external_id); + } + } + let current = graph.segments.load(); + graph.segments.swap(GraphSegmentList { + mutable: current.mutable.clone(), + immutable: segments, + }); + // Cross-tier delta edges re-enter the write buffer under their + // ORIGINAL edge ids (later streamed REMOVEEDGE records must resolve). + let installed = graph.segments.load(); + for (ek, src, dst, edge_type, created_lsn, weight, props) in delta_edges { + let src_key: crate::graph::types::NodeKey = slotmap::KeyData::from_ffi(src).into(); + let dst_key: crate::graph::types::NodeKey = slotmap::KeyData::from_ffi(dst).into(); + // Round-2 finding E: `add_edge_across_tiers_with_id`'s aliveness + // check only fires for RESIDENT endpoints, and the write buffer is + // empty at this point — a corrupted blob referencing a nonexistent + // node would otherwise install silently. Fail loud instead: a + // correctly-behaving master only exports delta edges whose + // endpoints are frozen into the segments shipped in this blob. + let endpoint_known = |nk: crate::graph::types::NodeKey| { + graph.write_buf.get_node(nk).is_some() + || installed + .immutable + .iter() + .any(|s| s.lookup_node(nk).is_some()) + }; + if !endpoint_known(src_key) || !endpoint_known(dst_key) { + tracing::warn!( + graph = %String::from_utf8_lossy(&name), + edge = ek, + src, + dst, + "replica graph sync: delta edge references node(s) absent from \ + installed segments — dropped (blob corrupt or master bug)" + ); + continue; + } + if graph + .write_buf + .add_edge_across_tiers_with_id( + ek, + src_key, + dst_key, + edge_type, + weight, + props, + created_lsn, + ) + .is_err() + { + tracing::warn!( + graph = %String::from_utf8_lossy(&name), + edge = ek, + "replica graph sync: cross-tier delta edge failed to install" + ); + } + } + // Copy-up tombstones LAST: they shadow rows in the just-installed + // segments (restore_dead_shadows copy-ups from graph.segments). + graph.restore_dead_shadows(&shadows); + } + // The installed graphs exist only in memory; the next checkpoint must + // re-materialize them (same contract as WAL replay). + if graph_count > 0 { + store.mark_dirty(); + } + Some(graph_count) +} + +/// Minimal bounds-checked reader over the blob. +struct Cursor<'a> { + data: &'a [u8], + pos: usize, +} + +impl<'a> Cursor<'a> { + fn take(&mut self, n: usize) -> Option<&'a [u8]> { + let end = self.pos.checked_add(n)?; + if end > self.data.len() { + return None; + } + let s = &self.data[self.pos..end]; + self.pos = end; + Some(s) + } + fn u8(&mut self) -> Option { + Some(self.take(1)?[0]) + } + fn u16(&mut self) -> Option { + #[allow(clippy::unwrap_used)] // take(2) guarantees the length + Some(u16::from_le_bytes(self.take(2)?.try_into().unwrap())) + } + fn u32(&mut self) -> Option { + #[allow(clippy::unwrap_used)] // take(4) guarantees the length + Some(u32::from_le_bytes(self.take(4)?.try_into().unwrap())) + } + fn u64(&mut self) -> Option { + #[allow(clippy::unwrap_used)] // take(8) guarantees the length + Some(u64::from_le_bytes(self.take(8)?.try_into().unwrap())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::command::graph::graph_write::label_to_id; + use smallvec::smallvec; + + fn seeded_store() -> GraphStore { + let mut store = GraphStore::new(); + store + .create_graph(Bytes::from_static(b"g1"), 64_000, 0) + .unwrap(); + let lsn = store.allocate_lsn(); + let g = store.get_graph_mut(b"g1").unwrap(); + let person = label_to_id(b"Person"); + let name_key = label_to_id(b"name"); + let a = g.write_buf.add_node( + smallvec![person], + smallvec![( + name_key, + crate::graph::types::PropertyValue::String(Bytes::from_static(b"alice")) + )], + None, + lsn, + ); + let b = g.write_buf.add_node( + smallvec![person], + smallvec![( + name_key, + crate::graph::types::PropertyValue::String(Bytes::from_static(b"bob")) + )], + None, + lsn, + ); + g.write_buf + .add_edge(a, b, label_to_id(b"KNOWS"), 1.0, None, lsn) + .unwrap(); + store + } + + #[test] + fn export_install_round_trip_preserves_nodes_edges_props() { + let mut master = seeded_store(); + let blob = export_graph_store(&mut master); + + let mut replica = GraphStore::new(); + // Pre-existing local graph must be dropped (authoritative replace). + replica + .create_graph(Bytes::from_static(b"stale"), 64_000, 0) + .unwrap(); + let installed = install_graph_store(&mut replica, &blob).expect("valid blob"); + assert_eq!(installed, 1); + assert!(replica.get_graph(b"stale").is_none(), "stale graph kept"); + + let g = replica.get_graph(b"g1").expect("g1 installed"); + let segments = g.segments.load(); + let total_nodes: u32 = segments.immutable.iter().map(|s| s.node_count()).sum(); + let total_edges: u32 = segments.immutable.iter().map(|s| s.edge_count()).sum(); + assert_eq!(total_nodes, 2); + assert_eq!(total_edges, 1); + // Property fidelity through the freeze → blob → install pipeline. + let name_key = label_to_id(b"name"); + let mut names: Vec = Vec::new(); + for seg in &segments.immutable { + for row in 0..seg.node_count() { + if let Some(crate::graph::types::PropertyValue::String(s)) = seg + .node_properties(row) + .iter() + .find(|(k, _)| *k == name_key) + .map(|(_, v)| v.clone()) + { + names.push(s); + } + } + } + names.sort(); + assert_eq!( + names, + vec![Bytes::from_static(b"alice"), Bytes::from_static(b"bob")] + ); + } + + /// Adversarial-review P1-5 + the write-buf remainder gaps found while + /// fixing it: (1) segment validity overlays (soft-deleted edges) are not + /// part of the CSR byte format, (2) cross-tier delta edges are RETAINED + /// by freeze() and live in no segment, (3) dead copy-up shadow nodes + /// tombstone frozen rows from the write buffer. All three exist only in + /// master memory + WAL — a replica must get them from the blob or it + /// resurrects deleted data and silently loses cross-tier edges. + #[test] + fn validity_overlay_delta_edges_and_dead_shadows_survive_export() { + use crate::graph::segment::GraphSegmentList; + use slotmap::Key; + + let mut master = GraphStore::new(); + master + .create_graph(Bytes::from_static(b"g1"), 64_000, 0) + .unwrap(); + let lsn = master.allocate_lsn(); + let g = master.get_graph_mut(b"g1").unwrap(); + let person = label_to_id(b"Person"); + let a = g + .write_buf + .add_node(smallvec![person], smallvec![], None, lsn); + let b = g + .write_buf + .add_node(smallvec![person], smallvec![], None, lsn); + let c = g + .write_buf + .add_node(smallvec![person], smallvec![], None, lsn); + g.write_buf + .add_edge(a, b, label_to_id(b"KNOWS"), 1.0, None, lsn) + .unwrap(); + assert!(g.freeze_and_compact(lsn + 1), "freeze must succeed"); + + // (1) Soft-delete the frozen KNOWS edge — overlay only. + { + let segs = g.segments.load(); + assert_eq!(segs.immutable.len(), 1); + let mut heap = + CsrSegment::from_bytes(&segs.immutable[0].to_bytes()).expect("round-trip"); + heap.mark_deleted(0); + g.segments.swap(GraphSegmentList { + mutable: segs.mutable.clone(), + immutable: vec![std::sync::Arc::new(CsrStorage::Heap(heap))], + }); + } + // (2) Cross-tier delta edges between frozen endpoints: one with + // weight+props (record payload), one default-weight (empty record). + let delta_ek = g + .write_buf + .add_edge_across_tiers( + a, + b, + label_to_id(b"LIKES"), + 2.5, + Some(smallvec![( + label_to_id(b"since"), + crate::graph::types::PropertyValue::Int(2020), + )]), + lsn + 2, + ) + .expect("delta edge"); + g.write_buf + .add_edge_across_tiers(b, a, label_to_id(b"SEES"), 1.0, None, lsn + 2) + .expect("default-weight delta edge"); + // (3) Copy-up delete of frozen node c (dead shadow tombstone). + assert!(g.copy_up_node(c), "copy-up of frozen c"); + assert!(g.write_buf.remove_node(c, lsn + 3), "shadow delete"); + + let blob = export_graph_store(&mut master); + let mut replica = GraphStore::new(); + assert_eq!(install_graph_store(&mut replica, &blob), Some(1)); + + let rg = replica.get_graph(b"g1").expect("g1 installed"); + let segs = rg.segments.load(); + assert_eq!(segs.immutable.len(), 1); + // (1) The frozen KNOWS edge stays deleted on the replica. + assert!( + !segs.immutable[0].is_valid(0), + "segment validity overlay lost — deleted edge resurrected" + ); + // (2) Both delta edges re-installed under their original ids. + let delta: Vec<_> = rg.write_buf.iter_edges().collect(); + assert_eq!(delta.len(), 2, "cross-tier delta edges lost"); + let (ek0, e0) = delta + .iter() + .find(|(_, e)| e.edge_type == label_to_id(b"LIKES")) + .expect("LIKES delta edge"); + assert_eq!(ek0.data().as_ffi(), delta_ek.data().as_ffi()); + assert_eq!(e0.weight, 2.5); + assert_eq!( + e0.properties.as_ref().and_then(|p| p + .iter() + .find(|(k, _)| *k == label_to_id(b"since")) + .map(|(_, v)| v.clone())), + Some(crate::graph::types::PropertyValue::Int(2020)), + ); + // (3) Node c stays tombstoned (dead shadow re-materialized). + let shadow = rg.write_buf.get_node(c).expect("shadow present"); + assert_ne!(shadow.deleted_lsn, u64::MAX, "shadow must be dead"); + } + + #[test] + fn empty_store_round_trips_and_truncated_blob_rejected() { + let mut empty = GraphStore::new(); + let blob = export_graph_store(&mut empty); + let mut replica = seeded_store(); + assert_eq!(install_graph_store(&mut replica, &blob), Some(0)); + assert!( + replica.get_graph(b"g1").is_none(), + "empty master snapshot must drop replica-local graphs" + ); + + let mut master = seeded_store(); + let full = export_graph_store(&mut master); + let mut target = GraphStore::new(); + assert_eq!( + install_graph_store(&mut target, &full[..full.len() - 3]), + None, + "truncated blob must be rejected" + ); + } +} diff --git a/src/replication/master.rs b/src/replication/master.rs index 6f33db7b3..41025bb16 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -446,6 +446,9 @@ async fn register_replica_with_shards( // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up // protocol is wired on the single-shard inline path only. registered: None, + // Cross-shard registration: the target shard's offset is + // owned by its own thread — the arm reads it at drain. + push_offset: None, }; let _ = prod.try_push(msg); } @@ -537,6 +540,9 @@ async fn register_replica_with_shards( // by the R2 PrepareReplicaSync redesign; the offset-reply catch-up // protocol is wired on the single-shard inline path only. registered: None, + // Cross-shard registration: the target shard's offset is + // owned by its own thread — the arm reads it at drain. + push_offset: None, }; let _ = prod.try_push(msg); } @@ -602,7 +608,6 @@ pub async fn handle_psync_inline_single_shard( mut stream: monoio::net::TcpStream, repl_state: Arc>, _shard_databases: Arc, - dispatch_tx: Rc>>>, replica_addr: std::net::SocketAddr, ) -> anyhow::Result<()> { use monoio::io::AsyncWriteRentExt; @@ -647,6 +652,16 @@ pub async fn handle_psync_inline_single_shard( // inside the RDB AND above snapshot_offset — re-delivered via // catch-up, double-applying non-idempotent commands (INCR). // + // This atomicity argument additionally requires that every local + // write advances the offset IN its own synchronous stretch — + // `record_local_write` appends the backlog bytes and moves the + // counter at write time (only the live replica try_send is + // deferred to the event-loop drain). If the advance were deferred + // too (the pre-review design queued backlog+offset+fanout as one + // message), a mutation already visible to this RDB capture could + // still be BELOW `total_offset()` here, land in the catch-up + // range, and double-apply — adversarial-review P0-2. + // // The RDB is generated inline by reading all databases on shard 0. // Hold read guards across the synchronous write to avoid any // Clone requirement on Database (the type intentionally is not @@ -685,6 +700,13 @@ pub async fn handle_psync_inline_single_shard( )) } }; + // v0.7 graph replication: whole-graph-store snapshot + // (frozen CSR segments + id cursors). ALWAYS written when + // the graph feature is on — an empty blob (0 graphs) tells + // the replica the master authoritatively has none. + #[cfg(feature = "graph")] + let graph_blob = + crate::replication::graph_sync::export_graph_store(&mut s.graph_store); let mut moon_aux: Vec<(&[u8], &[u8])> = Vec::new(); if let Some(ref v) = vec_defs { moon_aux @@ -693,6 +715,11 @@ pub async fn handle_psync_inline_single_shard( if let Some(ref t) = text_defs { moon_aux.push((crate::persistence::redis_rdb::MOON_AUX_TEXT_DEFS, &t[..])); } + #[cfg(feature = "graph")] + moon_aux.push(( + crate::persistence::redis_rdb::MOON_AUX_GRAPH_STORE, + &graph_blob[..], + )); crate::persistence::redis_rdb::write_rdb_refs_with_moon_aux( &refs, &moon_aux, @@ -720,7 +747,7 @@ pub async fn handle_psync_inline_single_shard( // replica channel. Reading the backlog BEFORE registering (the // old order) left a window where a write drained in between // reached neither leg — a silent, unlogged replica gap. - let reg = push_register_replica_inline(&repl_state, &dispatch_tx)?; + let reg = push_register_replica_inline(&repl_state)?; let reg_offset = reg .reg_rx .recv_async() @@ -728,8 +755,7 @@ pub async fn handle_psync_inline_single_shard( .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?; } PsyncDecision::PartialResync { from_offset } => { let response = format!("+CONTINUE {}\r\n", repl_id); @@ -737,7 +763,7 @@ pub async fn handle_psync_inline_single_shard( wr.map_err(|e| anyhow::anyhow!(e))?; // Same register-then-catch-up ordering as the FullResync arm. - let reg = push_register_replica_inline(&repl_state, &dispatch_tx)?; + let reg = push_register_replica_inline(&repl_state)?; let reg_offset = reg .reg_rx .recv_async() @@ -745,8 +771,7 @@ pub async fn handle_psync_inline_single_shard( .map_err(|_| anyhow::anyhow!("event loop dropped registration reply"))?; send_backlog_range(&mut stream, &backlog_slot, from_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?; } } Ok(()) @@ -815,9 +840,7 @@ struct InlineReplicaRegistration { #[cfg(feature = "runtime-monoio")] fn push_register_replica_inline( repl_state: &Arc>, - dispatch_tx: &Rc>>>, ) -> anyhow::Result { - use ringbuf::traits::Producer; use std::sync::atomic::Ordering; static NEXT_REPLICA_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); @@ -831,20 +854,31 @@ fn push_register_replica_inline( .read() .map(|g| g.backlog_capacity) .unwrap_or(crate::replication::state::DEFAULT_REPL_BACKLOG_SIZE); - let mut prods = dispatch_tx.borrow_mut(); - if let Some(prod) = prods.get_mut(0) { - let msg = crate::shard::dispatch::ShardMessage::RegisterReplica { - replica_id, - tx: tx.clone(), - backlog_capacity, - registered: Some(reg_tx), - }; - if prod.try_push(msg).is_err() { - anyhow::bail!("failed to push RegisterReplica onto shard 0 SPSC"); - } - } else { - anyhow::bail!("shard 0 producer missing"); - } + // The inline PSYNC task runs ON the owning shard's thread; the SPSC mesh + // has no self-loop (N·(N−1) skip-self — at shards=1 the producer Vec is + // EMPTY), so registration goes through the thread-local self queue the + // event loop drains alongside its SPSC consumers. + // + // The live-fanout start offset is captured HERE, at push time — NOT at + // drain time. Local writes advance the shard offset synchronously at + // write time (`record_local_write`), so a write that lands between this + // push and the drain has already moved the counter; a drain-time read + // would put it below `reg_offset` (delivered via backlog catch-up) while + // its `ReplicaLiveFanout` message — queued BEHIND this registration — + // also delivers it live: double-applied on the replica. The push-time + // offset keeps catch-up and live delivery disjoint for every interleave + // (see `RegisterReplica::push_offset`). + let push_offset = repl_state + .read() + .map(|g| g.total_offset()) + .map_err(|_| anyhow::anyhow!("replication state lock poisoned"))?; + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::RegisterReplica { + replica_id, + tx: tx.clone(), + backlog_capacity, + registered: Some(reg_tx), + push_offset: Some(push_offset), + }); Ok(InlineReplicaRegistration { replica_id, tx, @@ -863,10 +897,8 @@ async fn drain_replica_inline_single_shard( addr: std::net::SocketAddr, stream: monoio::net::TcpStream, repl_state: Arc>, - dispatch_tx: Rc>>>, ) -> anyhow::Result<()> { use monoio::io::AsyncWriteRentExt; - use ringbuf::traits::Producer; let InlineReplicaRegistration { replica_id, @@ -908,13 +940,10 @@ async fn drain_replica_inline_single_shard( if let Ok(mut rs) = repl_state.write() { rs.replicas.retain(|r| r.id != replica_id); } - { - let mut prods = dispatch_tx.borrow_mut(); - if let Some(prod) = prods.get_mut(0) { - let _ = prod - .try_push(crate::shard::dispatch::ShardMessage::UnregisterReplica { replica_id }); - } - } + // Same-thread → self queue (no self-SPSC exists; see push_register_replica_inline). + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::UnregisterReplica { + replica_id, + }); Ok(()) } diff --git a/src/replication/mod.rs b/src/replication/mod.rs index bc63f2903..09ba39bf2 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -1,5 +1,7 @@ pub mod apply; pub mod backlog; +#[cfg(feature = "graph")] +pub mod graph_sync; pub mod handshake; pub mod master; pub mod replica; diff --git a/src/replication/state.rs b/src/replication/state.rs index f068e4657..6135e3ef4 100644 --- a/src/replication/state.rs +++ b/src/replication/state.rs @@ -111,11 +111,25 @@ impl ReplicationState { /// partial resync. Capacity comes from `self.backlog_capacity` /// (`--repl-backlog-size`, default 1 MiB per shard). pub fn ensure_backlogs_allocated(&self) { - for slot in &self.per_shard_backlogs { + // Hint FIRST: any write racing this allocation that still sees the + // hint as false advances the offset without a backlog append, and the + // seed below (reading the offset AFTER that advance) re-aligns. At + // shards=1 both run on the same shard thread, so there is no race at + // all. (Multi-shard masters ride the R2 redesign.) + mark_fanout_active(); + for (shard_id, slot) in self.per_shard_backlogs.iter().enumerate() { let mut guard = slot.lock(); if guard.is_none() { - *guard = Some(crate::replication::backlog::ReplicationBacklog::new( + // Seed byte positions at the CURRENT shard offset — see + // `ReplicationBacklog::new_at`. + let offset = self + .shard_offsets + .get(shard_id) + .map(|o| o.load(Ordering::Relaxed)) + .unwrap_or(0); + *guard = Some(crate::replication::backlog::ReplicationBacklog::new_at( self.backlog_capacity, + offset, )); } } @@ -276,6 +290,29 @@ pub fn save_replication_state( Ok(()) } +/// Process-global "a replica has (ever) attached" hint. +/// +/// Write hot paths gate their replication-fanout serialization on this ONE +/// Relaxed load instead of taking `repl_state.read()` per command (the same +/// S3.5a rationale that gave READONLY its `is_replica_mirror` AtomicBool). +/// Set by `ensure_backlogs_allocated` (REPLCONF/PSYNC arrival) and never +/// cleared — once a replica has attached, the master keeps feeding the +/// backlog so partial resync stays possible, exactly like Redis. +static FANOUT_HINT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + +/// Mark the fanout hint (idempotent). See [`FANOUT_HINT`]. +#[inline] +pub fn mark_fanout_active() { + FANOUT_HINT.store(true, Ordering::Relaxed); +} + +/// Cheap hot-path predicate: has any replica ever begun attaching? +/// False ⇒ skip replication serialization/fan-out entirely. +#[inline] +pub fn fanout_hint_active() -> bool { + FANOUT_HINT.load(Ordering::Relaxed) +} + /// Load replication IDs from {dir}/replication.state. /// If file missing or malformed, generates new IDs and saves them. pub fn load_replication_state(dir: &std::path::Path) -> (String, String) { diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 22dcd2ae5..3a0071928 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -669,6 +669,7 @@ pub(super) fn try_handle_info( #[inline] pub(super) fn try_enforce_readonly( cmd: &[u8], + cmd_args: &[Frame], ctx: &ConnectionContext, responses: &mut Vec, ) -> bool { @@ -679,6 +680,19 @@ pub(super) fn try_enforce_readonly( return false; } if metadata::is_write(cmd) { + // GRAPH.QUERY is blanket-W in the metadata table because Cypher CAN + // write; a read-only MATCH/RETURN must still be served by a replica. + // Reuse the token-scan classifier the write dispatch path branches + // on — it can false-POSITIVE (blocks a weird read) but never + // false-negative (lets a write through). + #[cfg(feature = "graph")] + if cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") + && !crate::command::graph::is_cypher_write_query(cmd_args) + { + return false; + } + #[cfg(not(feature = "graph"))] + let _ = cmd_args; responses.push(Frame::Error(Bytes::from_static( b"READONLY You can't write against a read only replica.", ))); diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index 1000c3820..6adeba187 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -32,7 +32,12 @@ fn is_replicated_ft_def_mutation(cmd: &[u8], cmd_args: &[Frame]) -> bool { /// snapshot is captured, so an FT.* mutation racing a first-ever attach still /// fans out). Gates the serialize+SPSC round trip so servers that never /// replicate pay nothing — mirrors `wal_fanout_has_work` on the KV path. -fn replication_fanout_active(ctx: &ConnectionContext) -> bool { +pub(super) fn replication_fanout_active(ctx: &ConnectionContext) -> bool { + // Cheap first gate: one Relaxed load; false until the first replica ever + // begins attaching (REPLCONF/PSYNC → ensure_backlogs_allocated). + if !crate::replication::state::fanout_hint_active() { + return false; + } ctx.repl_state.as_ref().is_some_and(|rs| { rs.read().is_ok_and(|g| { !g.replicas.is_empty() @@ -43,6 +48,66 @@ fn replication_fanout_active(ctx: &ConnectionContext) -> bool { }) } +/// Record one successfully-executed local write in the replication plane. +/// +/// The backlog append AND the shard-offset advance happen HERE, synchronously, +/// in the same no-await stretch as the keyspace mutation itself. This is the +/// linchpin of snapshot consistency: the inline PSYNC task reads +/// `total_offset()` in the same synchronous block as its RDB capture, so with +/// a synchronous advance a mutation baked into the RDB always has its offset +/// counted — the backlog catch-up range `[snapshot_offset, reg_offset)` can +/// never re-deliver it (double-applying INCR/LPUSH on the replica). The +/// previous design deferred backlog+offset to the event-loop drain of a +/// queued message, which opened exactly that window. +/// +/// Only the live replica `try_send` is deferred (`ReplicaLiveFanout` on the +/// self queue): the replica sender list lives in the event loop's local +/// state. Exactly-once holds for every interleave with a registering replica +/// because `RegisterReplica.push_offset` is also captured at push time — a +/// write W and a registration R queued in either order agree on whether W is +/// below `reg_offset` (delivered via catch-up, fan-out message drains before +/// R registers) or at/above it (delivered live, catch-up excludes it). +/// +/// A backlog slot that is `None` skips the byte append but still advances the +/// offset — mirroring `wal_append_and_fanout`: the offset counter is the +/// source of truth, and a later lazy allocation seeds the backlog at the +/// current counter (`ReplicationBacklog::new_at`). +/// +/// ⚠ Monoio shard threads only (pushes to `shard::self_msg`) — callers are +/// all inside `handler_monoio`, which is `runtime-monoio`-gated. +pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) { + if let Some(rs) = ctx.repl_state.as_ref() { + if let Ok(g) = rs.read() { + if let Some(slot) = g.per_shard_backlogs.get(ctx.shard_id) { + if let Some(backlog) = slot.lock().as_mut() { + backlog.append(&bytes); + } + } + g.increment_shard_offset(ctx.shard_id, bytes.len() as u64); + } + } + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes }); +} + +/// Fail-loud marker for planes NOT yet wired into replication (round-2 +/// finding A): WS.* and MQ.* writes persist durably on the master but never +/// reach a replica — deterministic record forms + replica apply arms are the +/// task-#34 follow-up. Warn ONCE per process so operators running replicas +/// learn about the divergence at write time instead of at failover. No-op +/// (one Relaxed load) when no replica has ever attached. +pub(super) fn warn_unreplicated_plane(ctx: &ConnectionContext, cmd: &[u8]) { + use std::sync::atomic::{AtomicBool, Ordering}; + static WARNED: AtomicBool = AtomicBool::new(false); + if replication_fanout_active(ctx) && !WARNED.swap(true, Ordering::Relaxed) { + tracing::warn!( + command = %String::from_utf8_lossy(cmd), + "replication: WS.*/MQ.* writes are NOT replicated in v0.7 — a \ + replica will not see this plane (known limitation, further \ + occurrences not logged)" + ); + } +} + /// Handle FT.* commands. Returns `true` if the command was consumed. /// /// Caller should `continue` the frame loop when this returns `true`. @@ -797,28 +862,17 @@ pub(super) async fn try_handle_ft_command( }); // v0.7 R0.5: index-DEFINITION mutations must reach replicas. FT.* // executes here at the connection layer (never crosses the SPSC write - // path), so on success fan the ORIGINAL command bytes into the - // replication plane via ReplicateVerbatim — exact parity, no - // reconstruction. Single-shard only (multi-shard FT.* replication - // rides the R2 broadcast redesign). The replica applies it through - // the same ft_create/ft_dropindex/ft_config handlers. + // path), so on success record the ORIGINAL command bytes in the + // replication plane — exact parity, no reconstruction. Single-shard + // only (multi-shard FT.* replication rides the R2 broadcast + // redesign). The replica applies it through the same + // ft_create/ft_dropindex/ft_config handlers. if !matches!(response, Frame::Error(_)) && is_replicated_ft_def_mutation(cmd, cmd_args) && replication_fanout_active(ctx) { - use ringbuf::traits::Producer; let serialized = crate::persistence::aof::serialize_command(frame); - let mut producers = ctx.dispatch_tx.borrow_mut(); - if let Some(prod) = producers.get_mut(0) { - let msg = - crate::shard::dispatch::ShardMessage::ReplicateVerbatim { bytes: serialized }; - if prod.try_push(msg).is_err() { - tracing::warn!( - "replication: SPSC full, {} not fanned to replicas (replicas must full-resync)", - String::from_utf8_lossy(cmd) - ); - } - } + record_local_write(ctx, serialized); } let mut response = response; if let Some(ws_id) = conn.workspace_id.as_ref() { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 4a2311b51..c1036f04c 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -906,7 +906,7 @@ pub(crate) async fn handle_connection_sharded_monoio< { continue; } - if dispatch::try_enforce_readonly(cmd, ctx, &mut responses) { + if dispatch::try_enforce_readonly(cmd, cmd_args, ctx, &mut responses) { continue; } // MA12: Disk full enforcement @@ -1278,8 +1278,13 @@ pub(crate) async fn handle_connection_sharded_monoio< } } - // Pre-classify write commands for AOF + tracking - let is_write = if ctx.aof_pool.is_some() || conn.tracking_state.enabled { + // Pre-classify write commands for AOF + tracking + replication + // fan-out (the fanout hint is one Relaxed load, false until the + // first replica ever begins attaching). + let is_write = if ctx.aof_pool.is_some() + || conn.tracking_state.enabled + || crate::replication::state::fanout_hint_active() + { metadata::is_write(cmd) } else { false @@ -1324,13 +1329,27 @@ pub(crate) async fn handle_connection_sharded_monoio< // awaits the writer's fsync ack before responding to // the client. if matches!(response, Frame::Integer(1)) { - if let Some(ref pool) = ctx.aof_pool { + // v0.7 local-leg live replication — same contract as + // the main write leg: record (backlog+offset, sync) + // before any await, AOF leg does not double-advance + // the offset (lsn = 0). + let repl_active = ft::replication_fanout_active(ctx); + if repl_active || ctx.aof_pool.is_some() { let serialized = aof::serialize_command(&frame); - let lsn = aof::AofWriterPool::issue_append_lsn( - &ctx.repl_state, - ctx.shard_id, - serialized.len(), - ); + let lsn = if repl_active { + ft::record_local_write(ctx, serialized.clone()); + 0 + } else { + aof::AofWriterPool::issue_append_lsn( + &ctx.repl_state, + ctx.shard_id, + serialized.len(), + ) + }; + let Some(ref pool) = ctx.aof_pool else { + responses.push(response); + continue; + }; match pool.send_append_group(ctx.shard_id, lsn, serialized).await { // Always: durability confirmed by ONE // fsync_barrier per batch (resolve_local_leg_barrier @@ -1389,13 +1408,27 @@ pub(crate) async fn handle_connection_sharded_monoio< // — `:0` (key absent / dst exists w/o REPLACE) is a no-op. // H1: durable path awaits fsync under appendfsync=always. if matches!(response, Frame::Integer(1)) { - if let Some(ref pool) = ctx.aof_pool { + // v0.7 local-leg live replication — same contract + // as the main write leg (record backlog+offset + // synchronously before await; AOF leg does not + // double-advance, lsn = 0). + let repl_active = ft::replication_fanout_active(ctx); + if repl_active || ctx.aof_pool.is_some() { let serialized = aof::serialize_command(&frame); - let lsn = aof::AofWriterPool::issue_append_lsn( - &ctx.repl_state, - ctx.shard_id, - serialized.len(), - ); + let lsn = if repl_active { + ft::record_local_write(ctx, serialized.clone()); + 0 + } else { + aof::AofWriterPool::issue_append_lsn( + &ctx.repl_state, + ctx.shard_id, + serialized.len(), + ) + }; + let Some(ref pool) = ctx.aof_pool else { + responses.push(response); + continue; + }; match pool.send_append_group(ctx.shard_id, lsn, serialized).await { // Same one-barrier-per-batch contract as MOVE. Ok(true) => local_leg_write_idxs.push(responses.len()), @@ -1671,20 +1704,40 @@ pub(crate) async fn handle_connection_sharded_monoio< // measured 8x deficit vs Redis at P16). let mut aof_barrier_pending = false; if !matches!(response, Frame::Error(_)) && is_write { - if let Some(ref pool) = ctx.aof_pool { + // v0.7 local-leg live replication: record the wire + // bytes BEFORE any await — `record_local_write` does + // the backlog append + offset advance synchronously + // (mutation and replication record are one no-await + // stretch, atomic w.r.t. the inline PSYNC task's + // snapshot capture on this thread) and defers only + // the live replica try_send to the event-loop drain. + // The AOF leg below must NOT also advance the offset + // (lsn = 0; per-shard order is append order, same + // contract as wal_append_and_fanout's cross-shard + // legs). + let repl_active = ft::replication_fanout_active(ctx); + if repl_active || ctx.aof_pool.is_some() { let serialized = aof::serialize_command(&frame); - let lsn = aof::AofWriterPool::issue_append_lsn( - &ctx.repl_state, - ctx.shard_id, - serialized.len(), - ); - match pool.send_append_group(ctx.shard_id, lsn, serialized).await { - Ok(true) => aof_barrier_pending = true, - Ok(false) => {} - Err(_) => { - response = - Frame::Error(bytes::Bytes::from_static(aof::AOF_FSYNC_ERR)); - aof_failed = true; + let lsn = if repl_active { + ft::record_local_write(ctx, serialized.clone()); + 0 + } else { + aof::AofWriterPool::issue_append_lsn( + &ctx.repl_state, + ctx.shard_id, + serialized.len(), + ) + }; + if let Some(ref pool) = ctx.aof_pool { + match pool.send_append_group(ctx.shard_id, lsn, serialized).await { + Ok(true) => aof_barrier_pending = true, + Ok(false) => {} + Err(_) => { + response = Frame::Error(bytes::Bytes::from_static( + aof::AOF_FSYNC_ERR, + )); + aof_failed = true; + } } } } diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index 7c6e1a932..fbfb71cd2 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/src/server/conn/handler_monoio/txn.rs @@ -339,6 +339,24 @@ pub(super) async fn try_handle_temporal_invalidate( }); match result { Ok(()) => { + // v0.7 graph replication (round-2 finding B): + // TEMPORAL.INVALIDATE mutates graph state (valid_to) + // — stream the deterministic wall-clock-pinned form. + // 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). Same single- + // shard scope + synchronous-stretch contract as the + // GRAPH.* leg in write.rs. + if ctx.num_shards == 1 && super::ft::replication_fanout_active(ctx) { + let record = crate::command::temporal::serialize_invalidate_at( + &graph_name, + is_node, + entity_id, + wall_ms, + ); + super::ft::record_local_write(ctx, Bytes::from(record)); + } for record in wal_records { ctx.shard_databases .wal_append(ctx.shard_id, Bytes::from(record)); diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 7d1bc3554..d33e66c71 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -46,6 +46,16 @@ pub(super) async fn try_handle_ws_command( } }; + // Round-2 finding A fail-loud: WS.CREATE/WS.DROP persist locally + // (WorkspaceCreate/Drop WAL records) but are NOT replicated in v0.7 — + // surface the divergence once instead of letting a replica silently miss + // the plane. (WS.CREATE is non-deterministic — fresh UUIDv7 per execution + // — so verbatim streaming would be wrong; task #34 tracks the id-pinned + // record form.) + if sub.eq_ignore_ascii_case(b"CREATE") || sub.eq_ignore_ascii_case(b"DROP") { + super::ft::warn_unreplicated_plane(ctx, cmd); + } + if sub.eq_ignore_ascii_case(b"CREATE") { match validate_ws_create(cmd_args) { Ok(ws_name) => { @@ -301,6 +311,13 @@ pub(super) async fn try_handle_mq_command( } }; + // Round-2 finding A fail-loud: MQ mutations persist locally (WAL via + // execute_mq_on_owner) but are NOT replicated in v0.7 — surface the + // divergence once instead of letting a replica silently miss the plane. + if !sub.eq_ignore_ascii_case(b"LEN") && !sub.eq_ignore_ascii_case(b"DLQLEN") { + super::ft::warn_unreplicated_plane(ctx, cmd); + } + if sub.eq_ignore_ascii_case(b"CREATE") { match validate_mq_create(cmd_args) { Ok((queue_key, _max_delivery_count, _debounce_ms)) => { @@ -729,6 +746,29 @@ pub(super) async fn try_handle_multi_exec( &ctx.cached_clock, exec_publishes, ); + // v0.7 REPLICATION (adversarial-review P0-1): the txn body must + // reach replicas like any other successful local write. This was + // the ONE local write path that skipped the replication plane — + // `MULTI/SET/EXEC` at shards=1 committed on the master and never + // reached the replica (silent deterministic divergence). Record + // each body entry HERE, in the same synchronous stretch as the + // just-returned (fully synchronous) `execute_transaction_sharded` + // — atomic w.r.t. the inline PSYNC snapshot capture — and tell + // `persist_txn_aof` not to double-advance the offset (lsn = 0, + // same contract as the single-command legs). + let repl_active = super::ft::replication_fanout_active(ctx); + if repl_active { + // Round-2 finding G (throughput note): each entry pushes one + // ReplicaLiveFanout onto the self queue, and the drain + // preamble processes the whole burst before the shard's + // bounded SPSC consumers — a very large EXEC body is a tail- + // latency vector for cross-shard traffic sharing this thread. + // Each drain iteration is just a try_send per replica, so the + // burst is cheap; revisit only if EXEC bodies grow unbounded. + for bytes in &aof_entries { + super::ft::record_local_write(ctx, bytes.clone()); + } + } // DURABILITY: append every successful write in the body to THIS // shard's AOF via the same group-commit path as normal writes, then // issue ONE fsync barrier under appendfsync=always before acking. @@ -736,7 +776,7 @@ pub(super) async fn try_handle_multi_exec( // so ctx.shard_id is the correct AOF target. On barrier failure we // surface AOF_FSYNC_ERR instead of a false EXEC success — parity // with the normal write path. - if crate::server::conn::shared::persist_txn_aof(ctx, aof_entries) + if crate::server::conn::shared::persist_txn_aof(ctx, aof_entries, repl_active) .await .is_err() { @@ -928,9 +968,27 @@ pub(super) async fn try_handle_graph_command( txn.record_graph_undo(undo_op); } } + // v0.7 graph replication: the drained WAL records are the DETERMINISTIC, + // id-pinned form of this mutation (GRAPH.ADDNODE …, FNV- + // hashed u16 label/prop ids — `label_to_id` is stateless, so master and + // replica agree). Record them verbatim in the replication plane + // (`record_local_write`: backlog + offset synchronously, live fan-out at + // the next drain), which replicas replay via `GraphReplayCollector` + // without re-allocating ids. Only the replication legs — the WAL copy is + // the local `wal_append` below, so nothing double-logs. Single-shard + // scope, matching the R0/R0.5 FT.* leg (multi-shard graph replication + // rides the R2 broadcast redesign). + let wal_records: Vec = wal_records.into_iter().map(bytes::Bytes::from).collect(); + if !wal_records.is_empty() && ctx.num_shards == 1 && super::ft::replication_fanout_active(ctx) { + // Recording in the same synchronous stretch as the graph mutation + // keeps mutation + replication record atomic w.r.t. the inline PSYNC + // task's snapshot capture on this thread. + for record in &wal_records { + super::ft::record_local_write(ctx, record.clone()); + } + } for record in wal_records { - ctx.shard_databases - .wal_append(ctx.shard_id, bytes::Bytes::from(record)); + ctx.shard_databases.wal_append(ctx.shard_id, record); } let mut response = response; if let Some(ws_id) = conn.workspace_id.as_ref() { diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index ef0cba99f..41c5fbe31 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -562,6 +562,7 @@ pub(super) async fn try_handle_swapdb( #[inline] pub(super) fn try_enforce_readonly( cmd: &[u8], + cmd_args: &[Frame], ctx: &ConnectionContext, responses: &mut Vec, ) -> bool { @@ -572,6 +573,17 @@ pub(super) fn try_enforce_readonly( return false; } if metadata::is_write(cmd) { + // GRAPH.QUERY is blanket-W (Cypher CAN write); serve read-only + // MATCH/RETURN on replicas. The classifier never false-negatives + // for a write query — see handler_monoio::dispatch. + #[cfg(feature = "graph")] + if cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") + && !crate::command::graph::is_cypher_write_query(cmd_args) + { + return false; + } + #[cfg(not(feature = "graph"))] + let _ = cmd_args; responses.push(Frame::Error(Bytes::from_static( b"READONLY You can't write against a read only replica.", ))); diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index d246708be..bd66a5067 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -872,7 +872,7 @@ pub(crate) async fn handle_connection_sharded_inner< } // --- READONLY enforcement --- - if dispatch::try_enforce_readonly(cmd, ctx, &mut responses) { + if dispatch::try_enforce_readonly(cmd, cmd_args, ctx, &mut responses) { continue; } diff --git a/src/server/conn/handler_sharded/write.rs b/src/server/conn/handler_sharded/write.rs index 6ed786724..6fdc02590 100644 --- a/src/server/conn/handler_sharded/write.rs +++ b/src/server/conn/handler_sharded/write.rs @@ -683,7 +683,7 @@ pub(super) async fn try_handle_multi_exec( // so ctx.shard_id is the correct AOF target. On barrier failure we // surface AOF_FSYNC_ERR instead of a false EXEC success — parity // with the normal write path. - if crate::server::conn::shared::persist_txn_aof(ctx, aof_entries) + if crate::server::conn::shared::persist_txn_aof(ctx, aof_entries, false) .await .is_err() { diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index f3709a1b7..649b062bb 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -899,7 +899,15 @@ pub async fn handle_connection( rs_guard.role, crate::replication::state::ReplicationRole::Replica { .. } ) { - if metadata::is_write(cmd) { + // GRAPH.QUERY is blanket-W (Cypher CAN + // write); serve read-only MATCH/RETURN on + // replicas — see handler_monoio::dispatch. + #[cfg(feature = "graph")] + let graph_ro = cmd.eq_ignore_ascii_case(b"GRAPH.QUERY") + && !crate::command::graph::is_cypher_write_query(cmd_args); + #[cfg(not(feature = "graph"))] + let graph_ro = false; + if metadata::is_write(cmd) && !graph_ro { responses.push(Frame::Error(Bytes::from_static( b"READONLY You can't write against a read only replica.", ))); diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index 456d58012..5980832a7 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -340,9 +340,17 @@ pub(crate) fn execute_transaction_sharded( /// `AOF_FSYNC_ERR` instead of acking a durability it can't guarantee. A no-op /// (returns `Ok`) when AOF is disabled (`aof_pool` is `None`) or the body wrote /// nothing. +/// +/// `repl_recorded`: the caller already recorded every entry in the +/// replication plane (`record_local_write`, monoio local-leg fanout), which +/// advanced the shard offset — this AOF leg must then NOT advance it again +/// (lsn = 0; per-shard order is append order, same contract as the +/// single-command write legs). The tokio handler passes `false` (tokio-side +/// master fanout is not wired; monoio is the production replication runtime). pub(crate) async fn persist_txn_aof( ctx: &crate::server::conn::core::ConnectionContext, aof_entries: Vec, + repl_recorded: bool, ) -> Result<(), ()> { if aof_entries.is_empty() { return Ok(()); @@ -352,11 +360,15 @@ pub(crate) async fn persist_txn_aof( }; let mut barrier_pending = false; for bytes in aof_entries { - let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn( - &ctx.repl_state, - ctx.shard_id, - bytes.len(), - ); + let lsn = if repl_recorded { + 0 + } else { + crate::persistence::aof::AofWriterPool::issue_append_lsn( + &ctx.repl_state, + ctx.shard_id, + bytes.len(), + ) + }; match pool.send_append_group(ctx.shard_id, lsn, bytes).await { Ok(true) => barrier_pending = true, Ok(false) => {} diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs index 9a8634379..79850b187 100644 --- a/src/shard/conn_accept.rs +++ b/src/shard/conn_accept.rs @@ -757,7 +757,6 @@ pub(crate) fn spawn_monoio_connection( _hijacked_psync = true; let repl_state_clone = conn_ctx.repl_state.clone(); let shard_databases_clone = conn_ctx.shard_databases.clone(); - let dispatch_tx_clone = conn_ctx.dispatch_tx.clone(); let parsed_addr: std::net::SocketAddr = hp_peer .parse() .unwrap_or_else(|_| std::net::SocketAddr::from(([0, 0, 0, 0], 0))); @@ -771,7 +770,6 @@ pub(crate) fn spawn_monoio_connection( stream, rs, shard_databases_clone, - dispatch_tx_clone, parsed_addr, ).await { tracing::warn!("PSYNC handler exited: {}", e); diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 77b7f6728..4169c51dd 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -470,19 +470,31 @@ pub enum ShardMessage { /// replica gap). `None` = legacy fire-and-forget registration (the /// multi-shard paths, redesigned in R2). registered: Option>, + /// Live-fanout start offset captured by the pusher AT PUSH TIME, on + /// the shard's own thread (same-thread self-queue pushes only; `None` + /// for the cross-shard legacy registrations, where the arm replies + /// with the offset at drain). Same-thread pushes MUST set this: local + /// writes advance the shard offset synchronously at write time + /// (`record_local_write`), so an offset read at DRAIN could include a + /// write whose `ReplicaLiveFanout` message is queued BEHIND this + /// registration — the catch-up range would cover it AND the fan-out + /// message would deliver it live: double-applied on the replica. + push_offset: Option, }, /// Remove a replica's sender channel from this shard's fan-out list. /// Called when a replica disconnects or REPLICAOF NO ONE is executed. UnregisterReplica { replica_id: u64 }, - /// Fan a pre-serialized RESP command verbatim into the replication plane - /// (backlog + live replica streams + offset) WITHOUT touching WAL/AOF. + /// Deliver an already-RECORDED local write to the live replica streams. /// - /// For connection-layer commands that never cross the SPSC write path but - /// must replicate — FT.CREATE / FT.DROPINDEX / FT.CONFIG SET (v0.7 R0.5): - /// their durability is the vector/text sidecar, not the AOF, so only the - /// replication legs of `wal_append_and_fanout` apply. No-ops when no - /// replica has ever attached (no backlog, no replica_txs). - ReplicateVerbatim { bytes: bytes::Bytes }, + /// The producing thread (`replication::record_local_write`) has ALREADY + /// appended `bytes` to the shard backlog and advanced the shard offset, + /// synchronously with the keyspace mutation — so the inline PSYNC task's + /// snapshot capture can never observe a mutation whose offset is still + /// uncounted (that skew re-delivered the write via backlog catch-up, + /// double-applying non-idempotent commands on the replica). This message + /// carries ONLY the remaining leg: `try_send` to each registered + /// replica's sender channel. Same-thread self-queue only. + ReplicaLiveFanout { bytes: bytes::Bytes }, /// Register a CDC subscriber with this shard's fan-out registry (C3b-2). /// /// The connection handler creates a bounded channel, ships the sender diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index f7eaa524a..61a796bf6 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -766,6 +766,11 @@ impl super::Shard { let spsc_notify_local = spsc_notify; #[cfg(feature = "runtime-monoio")] let _ = &spsc_notify_local; + // Same-shard self-message queue (shard::self_msg): register this + // shard's drain Notify so a push from a sibling task on this thread + // (inline PSYNC registration, replication fan-out) wakes the drain + // arm immediately instead of waiting for the next periodic tick. + crate::shard::self_msg::register_drain_notify(spsc_notify_local.clone()); // tokio drains through the select! arms below and mutates the Vec // directly; monoio re-wraps it in Rc> for the spin probe. diff --git a/src/shard/mod.rs b/src/shard/mod.rs index a7aa5ad2c..4b5745d42 100644 --- a/src/shard/mod.rs +++ b/src/shard/mod.rs @@ -23,6 +23,7 @@ pub mod scatter_aggregate; pub mod scatter_hybrid; /// MA1: write-stall on immutable segment backlog. pub mod segment_stall; +pub mod self_msg; pub mod shared_databases; pub mod slice; pub mod spsc_handler; diff --git a/src/shard/self_msg.rs b/src/shard/self_msg.rs new file mode 100644 index 000000000..1f1ad1893 --- /dev/null +++ b/src/shard/self_msg.rs @@ -0,0 +1,71 @@ +//! Same-shard message queue — the self-loop the SPSC mesh doesn't have. +//! +//! `ChannelMesh` is N·(N−1) with skip-self mapping (`target_index` +//! debug-asserts `my_id != target_id`), so a task running ON a shard's own +//! thread cannot SPSC a `ShardMessage` to that shard. Before this module, +//! three same-thread producers silently no-op'd at `shards=1` (where the +//! producer Vec is EMPTY) and would have targeted the WRONG shard at +//! `shards>1`: +//! +//! 1. the inline PSYNC task's `RegisterReplica` (master.rs) — every replica +//! attach failed with "shard 0 producer missing" and the replica fell +//! into a 0.5s reconnect/full-resync loop that MASKED the dead live +//! stream (each resync's RDB carried the latest keyspace + FT defs); +//! 2. the FT.* index-definition replication fan-out (ft.rs); +//! 3. the graph WAL-record replication fan-out (write.rs). +//! +//! Replication messages carried here are DELIVERY-ONLY (`ReplicaLiveFanout`): +//! the backlog append and shard-offset advance happen synchronously at write +//! time in `record_local_write`, atomic with the mutation w.r.t. the inline +//! PSYNC task's snapshot capture. Deferring the offset advance to the drain +//! (the original design) let a mutation sit inside a FULLRESYNC RDB while +//! still below the advertised snapshot offset — re-delivered via backlog +//! catch-up, double-applying non-idempotent commands (adversarial-review +//! P0-2). `RegisterReplica.push_offset` is the matching pusher-side capture. +//! +//! One shard per OS thread (monoio thread-per-core), so a `thread_local!` +//! queue IS the per-shard self-channel — same pattern as `shard::slice`. +//! The event loop drains it inside `drain_spsc_shared` (ahead of the SPSC +//! consumers) and registers its `Notify` here at startup so a push from a +//! sibling task wakes a parked loop instead of waiting for the 1ms tick. +//! +//! ⚠ Tokio (work-stealing) tasks must NOT push here — their thread is not a +//! shard thread. All current producers are monoio-only paths. + +use std::cell::RefCell; +use std::collections::VecDeque; +use std::sync::Arc; + +use crate::runtime::channel::Notify; +use crate::shard::dispatch::ShardMessage; + +thread_local! { + static SELF_QUEUE: RefCell> = const { RefCell::new(VecDeque::new()) }; + static DRAIN_NOTIFY: RefCell>> = const { RefCell::new(None) }; +} + +/// Register the shard event loop's SPSC-drain `Notify` for this thread. +/// Called once at event-loop startup; a later `push` wakes the drain arm +/// immediately instead of stranding the message until the periodic tick. +pub fn register_drain_notify(notify: Arc) { + DRAIN_NOTIFY.with(|n| *n.borrow_mut() = Some(notify)); +} + +/// Enqueue a message for THIS shard's own drain loop and wake it. +/// +/// Caller must be on a shard thread (connection handler / task spawned by +/// the shard's event loop). Messages are handled by the same +/// `handle_shard_message_shared` the SPSC consumers feed, in FIFO order. +pub fn push(msg: ShardMessage) { + SELF_QUEUE.with(|q| q.borrow_mut().push_back(msg)); + DRAIN_NOTIFY.with(|n| { + if let Some(notify) = n.borrow().as_ref() { + notify.notify_one(); + } + }); +} + +/// Pop the next self-message (drain side; event-loop thread only). +pub fn pop() -> Option { + SELF_QUEUE.with(|q| q.borrow_mut().pop_front()) +} diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index 199abc109..c0d8772b7 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -172,6 +172,31 @@ pub(crate) fn drain_spsc_shared( execute_batch.clear(); other_messages.clear(); + // Self-queue FIRST: same-shard tasks (inline PSYNC RegisterReplica, + // local-write ReplicaLiveFanout) cannot SPSC to their own shard — the + // mesh is N·(N−1) skip-self — so they enqueue on the thread-local self + // queue (`shard::self_msg`). All self messages are control-plane arms + // (never Execute-batch / SnapshotBegin types), so they route through + // `other_messages`. FIFO here is a correctness invariant: a write's + // ReplicaLiveFanout drained before RegisterReplica never live-sends to + // the not-yet-registered replica (it gets those bytes via backlog + // catch-up instead), one drained after fans to the freshly-registered + // replica — no gap, no double-delivery (see RegisterReplica.push_offset). + // + // Drained UNBOUNDED, on purpose: every arm is cheap (try_send loop / + // vec push), the queue refills only while this thread's own tasks run + // (bounded by one loop iteration's frame budget), and an entry left + // behind by a MAX_DRAIN_PER_CYCLE cut-off would delay a replica's live + // bytes by up to a full tick. The SPSC consumers below keep their + // per-cycle bound. + loop { + let Some(msg) = crate::shard::self_msg::pop() else { + break; + }; + drained += 1; + other_messages.push(msg); + } + let mut snapshot_seen = false; for consumer in consumers.iter_mut() { if snapshot_seen { @@ -2477,6 +2502,7 @@ pub(crate) fn handle_shard_message_shared( tx, backlog_capacity, registered, + push_offset, } => { // Lazy-init replication backlog on first replica registration (saves 1MB/shard). // The backlog is shared with PSYNC handlers via Arc>> on @@ -2484,47 +2510,57 @@ pub(crate) fn handle_shard_message_shared( // earlier allocation point triggered by REPLCONF. Capacity is carried // in the message (from `--repl-backlog-size`) so this fallback can't // silently diverge from the handshake-path allocation. + crate::replication::state::mark_fanout_active(); let mut guard = repl_backlog.lock(); if guard.is_none() { - *guard = Some(ReplicationBacklog::new(backlog_capacity)); - } - drop(guard); - replica_txs.push((replica_id, tx)); - // Reply with the offset at which live fan-out begins. This runs - // synchronously between drains, so every fanout message queued - // BEFORE this registration has already advanced the offset, and - // every one after it will reach `tx` — the PSYNC task's catch-up - // read below this offset is therefore gap-free and overlap-free. - if let Some(reg_tx) = registered { + // Seed byte positions at the current shard offset so range + // math stays aligned with pre-attach `issue_lsn` advances — + // see `ReplicationBacklog::new_at`. let offset = repl_state .as_ref() .map(|h| h.shard_offset(shard_id)) .unwrap_or(0); + *guard = Some(ReplicationBacklog::new_at(backlog_capacity, offset)); + } + drop(guard); + replica_txs.push((replica_id, tx)); + // Reply with the offset at which live fan-out begins. For + // same-thread self-queue registrations this is `push_offset`, + // captured AT PUSH TIME: local writes advance the offset + // synchronously at write time (`record_local_write`), so a write + // W that lands between the registration's push and this drain has + // already advanced the counter — an offset read HERE would cover + // W in the catch-up range while W's `ReplicaLiveFanout` message + // (queued behind this registration, drained after it) ALSO + // delivers it live: double-applied. With the push-time offset the + // ledger is exact either way W interleaves: W before the push → + // below `reg_offset`, delivered via catch-up, its fan-out message + // drains before `tx` is registered (no live copy); W after the + // push → at/above `reg_offset`, excluded from catch-up, delivered + // live. Cross-shard legacy registrations (`None`, R2 redesign) + // keep the drain-time read. + if let Some(reg_tx) = registered { + let offset = push_offset.unwrap_or_else(|| { + repl_state + .as_ref() + .map(|h| h.shard_offset(shard_id)) + .unwrap_or(0) + }); let _ = reg_tx.send(offset); } } ShardMessage::UnregisterReplica { replica_id } => { replica_txs.retain(|(id, _)| *id != replica_id); } - ShardMessage::ReplicateVerbatim { bytes } => { - // Replication legs only: backlog + offset + live replica fan-out. - // WAL writer / AOF pool are deliberately None — the commands routed - // here (FT.CREATE / FT.DROPINDEX / FT.CONFIG SET) are durable via - // the vector/text sidecars, and an AOF copy would double-apply on - // recovery. `wal_fanout_has_work` no-ops the whole call when no - // replica has ever attached. - let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND; - let _ = wal_append_and_fanout( - &bytes, - &mut None, - repl_backlog, - replica_txs, - repl_state, - shard_id, - None, - false, - &mut aof_budget, - ); + ShardMessage::ReplicaLiveFanout { bytes } => { + // Live-delivery leg ONLY: backlog append + offset advance already + // happened synchronously at write time on this same thread + // (`record_local_write`) — doing either again here would double- + // count. Lagging replicas are skipped (try_send), same policy as + // `wal_append_and_fanout`'s fan-out leg. + for (_id, tx) in replica_txs.iter() { + let _ = tx.try_send(bytes.clone()); + } } ShardMessage::MigrateConnection(_) => { // MigrateConnection is collected by drain_spsc_shared into pending_migrations, diff --git a/tests/replication_graph.rs b/tests/replication_graph.rs new file mode 100644 index 000000000..7a28fe738 --- /dev/null +++ b/tests/replication_graph.rs @@ -0,0 +1,378 @@ +//! v0.7 graph-plane replication: a replica must synchronize graph data +//! (nodes, edges, labels, properties) from its master — both the live +//! GRAPH.* mutation stream and the PSYNC full-resync snapshot backfill. +//! +//! Single-shard master scope (matches R0/R0.5). Graph mutations are streamed +//! as their DETERMINISTIC, id-pinned WAL records (GRAPH.ADDNODE …, +//! GRAPH.ADDEDGE …) — the master's WAL layer already +//! rewrites non-deterministic Cypher into these, so the replica replays them +//! into an identical graph without re-allocating ids. +//! +//! Run: `MOON_BIN=./target/release/moon cargo test --test replication_graph -- --ignored` + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +fn moon_bin() -> String { + std::env::var("MOON_BIN").unwrap_or_else(|_| "./target/release/moon".to_string()) +} + +fn start_moon(port: u16, dir: &str) -> Child { + start_moon_logged(port, dir, None) +} + +/// Like `start_moon`, optionally teeing stderr (tracing output) to a file so +/// tests can assert stream-health properties — e.g. that the replica held ONE +/// live connection instead of masking a dead stream behind a 0.5s +/// reconnect/full-resync loop (the exact failure mode that hid the missing +/// self-SPSC producer for months). +fn start_moon_logged(port: u16, dir: &str, stderr_log: Option<&std::path::Path>) -> Child { + let stderr = match stderr_log { + Some(p) => Stdio::from(std::fs::File::create(p).expect("create log file")), + None => Stdio::null(), + }; + Command::new(moon_bin()) + .env("RUST_LOG", "moon=warn") + .args([ + "--port", + &port.to_string(), + "--shards", + "1", + "--dir", + dir, + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + ]) + .stdout(Stdio::null()) + .stderr(stderr) + .spawn() + .expect("Failed to start moon (set MOON_BIN to a built binary)") +} + +/// Kill-on-drop guard so a panicking assertion never leaks a live server. +struct Killer(Child); +impl Drop for Killer { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// Send one inline command, return the raw reply text. +fn send_cmd(addr: &str, cmd: &str) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream.set_read_timeout(Some(Duration::from_secs(5))).ok(); + if stream.write_all(format!("{cmd}\r\n").as_bytes()).is_err() { + return String::new(); + } + read_quiet(&mut stream) +} + +/// RESP array of bulk strings — binary-safe, for multiword graph commands. +fn send_resp(addr: &str, parts: &[&str]) -> String { + let Ok(mut stream) = TcpStream::connect(addr) else { + return String::new(); + }; + stream + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let mut out = format!("*{}\r\n", parts.len()).into_bytes(); + for p in parts { + out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes()); + out.extend_from_slice(p.as_bytes()); + out.extend_from_slice(b"\r\n"); + } + if stream.write_all(&out).is_err() { + return String::new(); + } + read_quiet(&mut stream) +} + +fn read_quiet(stream: &mut TcpStream) -> String { + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let deadline = Instant::now() + Duration::from_millis(600); + while Instant::now() < deadline { + match stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + Err(_) => { + if !buf.is_empty() { + break; + } + } + } + } + String::from_utf8_lossy(&buf).into_owned() +} + +fn wait_until bool>(timeout: Duration, f: F) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if f() { + return true; + } + std::thread::sleep(Duration::from_millis(150)); + } + f() +} + +/// REPL-GRAPH-01: live GRAPH.* mutations on the master reach the replica. +/// The replica attaches to an EMPTY master, then every subsequent graph write +/// must appear — isolating the live streaming leg from snapshot backfill. +#[test] +#[ignore] +fn replica_syncs_graph_live_stream() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + let master_addr = "127.0.0.1:16730"; + let replica_addr = "127.0.0.1:16731"; + + let replica_log = replica_dir.path().join("replica-stderr.log"); + + let _master = Killer(start_moon(16730, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + let _replica = Killer(start_moon_logged( + 16731, + replica_dir.path().to_str().unwrap(), + Some(&replica_log), + )); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + assert!( + send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16730").starts_with("+OK"), + "REPLICAOF failed" + ); + // Let the handshake settle before the first write. + std::thread::sleep(Duration::from_millis(500)); + + // 1. Live graph creation + node inserts stream to the replica. + assert!(send_cmd(master_addr, "GRAPH.CREATE g1").contains("OK")); + let a = send_resp( + master_addr, + &["GRAPH.ADDNODE", "g1", "Person", "name", "alice"], + ); + let b = send_resp( + master_addr, + &["GRAPH.ADDNODE", "g1", "Person", "name", "bob"], + ); + assert!(a.starts_with(':'), "ADDNODE alice must return an id: {a}"); + assert!(b.starts_with(':'), "ADDNODE bob must return an id: {b}"); + + let counted = wait_until(Duration::from_secs(10), || { + let r = send_resp( + replica_addr, + &["GRAPH.QUERY", "g1", "MATCH (n:Person) RETURN count(n)"], + ); + r.contains('2') + }); + assert!( + counted, + "replica never saw the 2 streamed Person nodes: {}", + send_resp( + replica_addr, + &["GRAPH.QUERY", "g1", "MATCH (n:Person) RETURN count(n)"] + ) + ); + + // 2. Node PROPERTIES survive replication (not just node count). + let names = send_resp( + replica_addr, + &["GRAPH.QUERY", "g1", "MATCH (n:Person) RETURN n.name"], + ); + assert!( + names.contains("alice") && names.contains("bob"), + "replica node properties diverged: {names}" + ); + + // 3. A brand-new graph created live shows up in the replica's GRAPH.LIST. + assert!(send_cmd(master_addr, "GRAPH.CREATE g2").contains("OK")); + let listed = wait_until(Duration::from_secs(10), || { + send_cmd(replica_addr, "GRAPH.LIST").contains("g2") + }); + assert!( + listed, + "replica GRAPH.LIST never listed live-created g2: {}", + send_cmd(replica_addr, "GRAPH.LIST") + ); + + // 4. STREAM-HEALTH: everything above must have arrived on ONE live + // stream. A reconnect/full-resync loop can deliver the same data via + // repeated snapshots (each FULLRESYNC RDB carries the graph aux) and + // silently mask a dead live stream — the exact failure mode that hid the + // missing self-SPSC producer behind green tests. + let log = std::fs::read_to_string(&replica_log).unwrap_or_default(); + let reconnects = log.matches("reconnecting").count(); + assert_eq!( + reconnects, 0, + "replica reconnected {reconnects}x — live stream did not hold:\n{log}" + ); +} + +/// REPL-GRAPH-02: a replica attaching to a master that ALREADY has graph data +/// must backfill it from the PSYNC full-resync snapshot (not just live deltas). +#[test] +#[ignore] +fn replica_backfills_graph_from_snapshot() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + let master_addr = "127.0.0.1:16732"; + let replica_addr = "127.0.0.1:16733"; + + let _master = Killer(start_moon(16732, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + // Build graph state on the master BEFORE any replica exists. + assert!(send_cmd(master_addr, "GRAPH.CREATE gsnap").contains("OK")); + for name in ["alice", "bob", "carol"] { + let r = send_resp( + master_addr, + &["GRAPH.ADDNODE", "gsnap", "Person", "name", name], + ); + assert!(r.starts_with(':'), "seed ADDNODE {name} failed: {r}"); + } + + // Now attach a fresh replica — it must learn gsnap + its 3 nodes. + let _replica = Killer(start_moon(16733, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + assert!( + send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16732").starts_with("+OK"), + "REPLICAOF failed" + ); + + let backfilled = wait_until(Duration::from_secs(10), || { + send_resp( + replica_addr, + &["GRAPH.QUERY", "gsnap", "MATCH (n:Person) RETURN count(n)"], + ) + .contains('3') + }); + assert!( + backfilled, + "replica never backfilled the 3 pre-existing snapshot nodes: {}", + send_resp( + replica_addr, + &["GRAPH.QUERY", "gsnap", "MATCH (n:Person) RETURN count(n)"] + ) + ); + + // A live write AFTER the snapshot must also land (stream continues past + // the backfill boundary). + let r = send_resp( + master_addr, + &["GRAPH.ADDNODE", "gsnap", "Person", "name", "dave"], + ); + assert!(r.starts_with(':'), "post-snapshot ADDNODE failed: {r}"); + let grew = wait_until(Duration::from_secs(10), || { + send_resp( + replica_addr, + &["GRAPH.QUERY", "gsnap", "MATCH (n:Person) RETURN count(n)"], + ) + .contains('4') + }); + assert!(grew, "post-snapshot live node did not replicate"); +} + +/// REPL-GRAPH-03 (adversarial round-2 finding B): TEMPORAL.INVALIDATE mutates +/// graph state (`valid_to`) and must reach the replica — streamed as the +/// deterministic wall-clock-pinned `TEMPORAL.INVALIDATE-AT` form so master +/// and replica agree on the exact `valid_to`. Observable only through an +/// explicit `VALID_AT` filter: far-future is inside the default +/// `valid_to = i64::MAX` before invalidation and outside `valid_to = now` +/// after it. +#[test] +#[ignore] +fn replica_applies_temporal_invalidate() { + // ~ year 2255 in Unix ms: beyond "now", far short of i64::MAX. + const FAR_FUTURE_MS: &str = "9000000000000"; + + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + let master_addr = "127.0.0.1:16660"; + let replica_addr = "127.0.0.1:16661"; + + let _master = Killer(start_moon(16660, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + let _replica = Killer(start_moon(16661, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + assert!( + send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16660").starts_with("+OK"), + "REPLICAOF failed" + ); + std::thread::sleep(Duration::from_millis(500)); + + assert!(send_cmd(master_addr, "GRAPH.CREATE tg").contains("OK")); + let reply = send_resp( + master_addr, + &["GRAPH.ADDNODE", "tg", "Person", "name", "alice"], + ); + assert!(reply.starts_with(':'), "ADDNODE must return an id: {reply}"); + let node_id = reply.trim_start_matches(':').trim().to_string(); + + let visible_query = [ + "GRAPH.QUERY", + "tg", + "MATCH (n:Person) RETURN n.name", + "VALID_AT", + FAR_FUTURE_MS, + ]; + assert!( + wait_until(Duration::from_secs(10), || send_resp( + replica_addr, + &visible_query + ) + .contains("alice")), + "node never became visible on the replica before invalidation" + ); + + let inv = send_resp( + master_addr, + &["TEMPORAL.INVALIDATE", &node_id, "NODE", "tg"], + ); + assert!(inv.starts_with("+OK"), "TEMPORAL.INVALIDATE failed: {inv}"); + // Master-side sanity: invalidation is observable at VALID_AT far-future. + assert!( + !send_resp(master_addr, &visible_query).contains("alice"), + "master itself still shows the node past its valid_to" + ); + + // The replica must converge to the same temporal visibility. + assert!( + wait_until(Duration::from_secs(10), || !send_resp( + replica_addr, + &visible_query + ) + .contains("alice")), + "TEMPORAL.INVALIDATE never applied on the replica (valid_to diverged): {}", + send_resp(replica_addr, &visible_query) + ); +} diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs index 99f6643e5..d6292ffe6 100644 --- a/tests/replication_streaming.rs +++ b/tests/replication_streaming.rs @@ -607,3 +607,84 @@ fn replica_attach_races_live_ft_create() { "replica lost FT.CREATE(s) during attach race.\nmaster: {master_list}\nreplica: {replica_list}" ); } + +/// REPL-STREAM-04 (adversarial-review P0-1 on the self-SPSC fix): MULTI/EXEC +/// bodies must replicate. +/// +/// The single-command local-write path records every successful write in the +/// replication plane, but EXEC persisted its body through `persist_txn_aof`'s +/// AOF-only leg: a `MULTI / SET / INCR / EXEC` on a `--shards 1` master with +/// an attached replica committed durably on the master and NEVER reached the +/// replica — no backlog bytes, no offset advance, no live fan-out. Silent, +/// deterministic divergence for every application using transactions. +#[test] +#[ignore] +fn replica_applies_multi_exec_bodies() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + + let master_addr = "127.0.0.1:16730"; + let replica_addr = "127.0.0.1:16731"; + + let _master = Killer(start_moon(16730, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + let _replica = Killer(start_moon(16731, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + + send_cmd(replica_addr, &format!("REPLICAOF 127.0.0.1 {}", 16730)); + + // Prove the live stream is up with a plain single write first. + send_cmd(master_addr, "SET plain alive"); + assert!( + wait_until(Duration::from_secs(10), || { + get(replica_addr, "plain").as_deref() == Some("alive") + }), + "single-command live stream not flowing — txn assertions would be meaningless" + ); + + // The transaction under test: a SET and two INCRs (INCR doubles as a + // double-apply canary — a re-delivered body would show ctr=4). + let exec_reply = send_seq( + master_addr, + &["MULTI", "SET t1 v1", "INCR ctr", "INCR ctr", "EXEC"], + ); + assert!( + !exec_reply.contains("ERR"), + "EXEC failed on the master: {exec_reply}" + ); + assert_eq!( + get(master_addr, "ctr").as_deref(), + Some("2"), + "master must see the txn's own effects" + ); + + // In-order stream sentinel: once this single post-txn write is visible, + // the txn body (streamed before it) must already have been applied. + send_cmd(master_addr, "SET txn_done 1"); + assert!( + wait_until(Duration::from_secs(10), || { + get(replica_addr, "txn_done").as_deref() == Some("1") + }), + "post-txn sentinel never replicated" + ); + + assert_eq!( + get(replica_addr, "t1").as_deref(), + Some("v1"), + "MULTI/EXEC SET did not replicate" + ); + assert_eq!( + get(replica_addr, "ctr").as_deref(), + Some("2"), + "MULTI/EXEC INCRs did not replicate exactly once (None=lost, 4=double-applied)" + ); +}