reset replica when injector detects potential corruption

libsql-replication/src/frame.rs

@@ -104,6 +104,12 @@ impl From<FrameBorrowed> for FrameMut {
     }
 }
 
+impl From<Box<FrameBorrowed>> for FrameMut {
+    fn from(inner: Box<FrameBorrowed>) -> Self {
+        Self { inner }
+    }
+}
+
 impl Frame {
     pub fn from_parts(header: &FrameHeader, data: &[u8]) -> Self {
         FrameBorrowed::from_parts(header, data).into()
@@ -134,6 +140,10 @@ impl FrameBorrowed {
         &self.page
     }
 
+    pub fn page_mut(&mut self) -> &mut [u8] {
+        &mut self.page
+    }
+
     pub fn header(&self) -> &FrameHeader {
         &self.header
     }

libsql-server/Cargo.toml

@@ -77,7 +77,7 @@ tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 http-body = "0.4"
 url = { version = "2.3", features = ["serde"] }
 uuid = { version = "1.3", features = ["v4", "serde"] }
-zerocopy = { version = "0.7.28", features = ["derive"] }
+zerocopy = { version = "0.7.28", features = ["derive", "alloc"] }
 
 [dev-dependencies]
 arbitrary = { version = "1.3.0", features = ["derive_arbitrary"] }

libsql-server/src/namespace/mod.rs

@@ -899,6 +899,11 @@ impl Namespace<ReplicaDatabase> {
                         (reset)(ResetOp::Destroy(namespace.clone()));
                         Err(err)?;
                     }
+                    e @ Error::Injector(_) => {
+                        tracing::error!("potential corruption detected while replicating, reseting  replica: {e}");
+                        (reset)(ResetOp::Reset(namespace.clone()));
+                        Err(e)?;
+                    },
                     Error::Meta(err) => {
                         use libsql_replication::meta::Error;
                         match err {
@@ -917,7 +922,6 @@ impl Namespace<ReplicaDatabase> {
                         }
                     }
                     e @ (Error::Internal(_)
-                    | Error::Injector(_)
                     | Error::Client(_)
                     | Error::PrimaryHandshakeTimeout
                     | Error::NeedSnapshot) => {

libsql-server/tests/cluster/replication.rs

@@ -1,5 +1,5 @@
-use std::sync::Arc;
 use std::time::Duration;
+use std::sync::Arc;
 
 use insta::assert_debug_snapshot;
 use libsql::Database;
@@ -14,7 +14,7 @@ use crate::common::{
 /// In this test, we first create a primary with a very small max_log_size, and then add a good
 /// amount of data to it. This will cause the primary to create a bunch of snaphots a large enough
 /// to prevent the replica from applying them all at once. We then start the replica, and check
-/// that it replicates correctly to the primary's replicaton index.
+/// that it replicates correctly to the primary's replicaton index. #[test]
 #[test]
 fn apply_partial_snapshot() {
     let mut sim = turmoil::Builder::new()