fix(replication): unified poison-record policy — decode failure kicks the link, never silent-skips (task #48)

CHANGELOG.md

@@ -6,6 +6,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — unified replica poison-record policy across graph/MQ/WS/temporal planes (task #48)
+
+Each replica-apply plane (graph WAL replay, MQ effect records, workspace
+create/drop, temporal invalidate, RESP framing) previously handled a
+malformed/undecodable record ad hoc — mostly `tracing::warn!` + silently
+skip-and-continue, one path (`WS.CREATE.APPLY`/`WS.DROP.APPLY`) even
+swallowed the failure entirely (`try_with_shard(...).is_some()` returned
+`true` regardless of the inner decode outcome). A silent skip means every
+subsequent record in the stream keeps applying against state that has
+already diverged from the master — invisible until an operator notices
+missing data.
+
+Unified policy (`replication::apply`, see its "Unified poison-record policy"
+module docs): any decode/parse failure on a live-stream record is now a
+`ApplyOutcome::Poisoned` — logged loud (rate-limited to 1/sec), counted in
+a new INFO counter, and propagated up so the replica connection task drops
+the link and lets the existing reconnect/resync loop renegotiate PSYNC
+(never silently skip, never panic). PSYNC snapshot install already
+fail-closed correctly (a malformed aux blob fails the whole install); it now
+also increments the same counter. Semantic apply errors on well-formed
+records (e.g. `WRONGTYPE`, "entity not found") keep the existing
+warn-and-continue posture — that is data-level divergence from a command
+that legitimately executed differently, not stream corruption.
+
+- New `INFO replication` field: `replication_poison_records_total`.
+- `src/replication/apply.rs`: `ApplyOutcome` enum, `poison()` helper,
+  `apply_graph`/`apply_mq`/`apply_ws_create`/`apply_ws_drop`/
+  `apply_temporal_invalidate` now return `bool` (poisoned vs. applied)
+  instead of logging-and-swallowing.
+- `src/replication/replica.rs`: both tokio and monoio stream-apply loops
+  match on `ApplyOutcome` and drop the connection on `Poisoned`, same as the
+  existing `NoShardSlice` / `DrainResult::fatal` paths.
 ### Added — real SHUTDOWN [NOSAVE|SAVE] (task #27)
 
 `SHUTDOWN` was previously a stub that always replied `ERR Errors trying to

src/command/connection.rs

@@ -304,12 +304,21 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
     // # Replication
     sections.push_str("# Replication\r\n");
     let (role, slaves, offset, repl_id) = crate::admin::metrics_setup::get_replication_info();
+    // task #48: unified poison-record counter, incremented by every replica
+    // apply plane (RESP framing / graph / MQ / WS / temporal / snapshot
+    // install) on a malformed/undecodable record — see
+    // `replication::apply`'s "Unified poison-record policy" docs. Zero on a
+    // master (a master never applies a replicated stream) and on a replica
+    // that has never seen a corrupt record.
+    let poison_total = crate::replication::apply::REPL_POISON_RECORDS_TOTAL
+        .load(std::sync::atomic::Ordering::Relaxed);
     let _ = write!(
         sections,
         "role:{role}\r\n\
          connected_slaves:{slaves}\r\n\
          master_replid:{repl_id}\r\n\
-         master_repl_offset:{offset}\r\n",
+         master_repl_offset:{offset}\r\n\
+         replication_poison_records_total:{poison_total}\r\n",
     );
     sections.push_str("\r\n");