introduce libsql injector

Author: MarinPostma

Cargo.lock

@@ -45,6 +45,7 @@ rusqlite = { package = "libsql-rusqlite", path = "vendored/rusqlite", version =
 ] }
 hyper = { version = "0.14" }
 tower = { version = "0.4.13" }
+zerocopy = { version = "0.7.32", features = ["derive", "alloc"] }
 
 # Config for 'cargo dist'
 [workspace.metadata.dist]

Cargo.toml

@@ -12,6 +12,7 @@ license = "MIT"
 tonic = { version = "0.11", features = ["tls"] }
 prost = "0.12"
 libsql-sys = { version = "0.7", path = "../libsql-sys", default-features = false, features = ["wal", "rusqlite", "api"] }
+libsql-wal = { path = "../libsql-wal/" }
 rusqlite = { workspace = true }
 parking_lot = "0.12.1"
 bytes = { version = "1.5.0", features = ["serde"] } 

libsql-replication/Cargo.toml

@@ -9,6 +9,12 @@ message LogOffset {
 
 message HelloRequest {
     optional uint64 handshake_version = 1;
+    enum WalFlavor {
+        Sqlite = 0;
+        Libsql = 1;
+    }
+    // the type of wal that the client is expecting
+    optional WalFlavor wal_flavor = 2;
 }
 
 message HelloResponse {

libsql-replication/proto/replication_log.proto

@@ -13,7 +13,6 @@ use crate::LIBSQL_PAGE_SIZE;
 pub type FrameNo = u64;
 
 /// The file header for the WAL log. All fields are represented in little-endian ordering.
-/// See `encode` and `decode` for actual layout.
 // repr C for stable sizing
 #[repr(C)]
 #[derive(Debug, Clone, Copy, zerocopy::FromZeroes, zerocopy::FromBytes, zerocopy::AsBytes)]
@@ -22,7 +21,7 @@ pub struct FrameHeader {
     pub frame_no: lu64,
     /// Rolling checksum of all the previous frames, including this one.
     pub checksum: lu64,
-    /// page number, if frame_type is FrameType::Page
+    /// page number
     pub page_no: lu32,
     /// Size of the database (in page) after committing the transaction. This is passed from sqlite,
     /// and serves as commit transaction boundary

libsql-replication/src/frame.rs

@@ -10,6 +10,48 @@ pub struct LogOffset {
 pub struct HelloRequest {
     #[prost(uint64, optional, tag = "1")]
     pub handshake_version: ::core::option::Option<u64>,
+    /// the type of wal that the client is expecting
+    #[prost(enumeration = "hello_request::WalFlavor", optional, tag = "2")]
+    pub wal_flavor: ::core::option::Option<i32>,
+}
+/// Nested message and enum types in `HelloRequest`.
+pub mod hello_request {
+    #[derive(
+        Clone,
+        Copy,
+        Debug,
+        PartialEq,
+        Eq,
+        Hash,
+        PartialOrd,
+        Ord,
+        ::prost::Enumeration
+    )]
+    #[repr(i32)]
+    pub enum WalFlavor {
+        Sqlite = 0,
+        Libsql = 1,
+    }
+    impl WalFlavor {
+        /// String value of the enum field names used in the ProtoBuf definition.
+        ///
+        /// The values are not transformed in any way and thus are considered stable
+        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+        pub fn as_str_name(&self) -> &'static str {
+            match self {
+                WalFlavor::Sqlite => "Sqlite",
+                WalFlavor::Libsql => "Libsql",
+            }
+        }
+        /// Creates an enum from field names used in the ProtoBuf definition.
+        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+            match value {
+                "Sqlite" => Some(Self::Sqlite),
+                "Libsql" => Some(Self::Libsql),
+                _ => None,
+            }
+        }
+    }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

libsql-replication/src/generated/wal_log.rs

@@ -1,11 +1,12 @@
 pub type Result<T, E=Error> = std::result::Result<T, E>;
+pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
     #[error("SQLite error: {0}")]
     Sqlite(#[from] rusqlite::Error),
-    #[error("A fatal error occured injecting frames")]
-    FatalInjectError,
+    #[error("A fatal error occured injecting frames: {0}")]
+    FatalInjectError(BoxError),
 }

libsql-replication/src/injector/error.rs

@@ -0,0 +1,44 @@
+use std::mem::size_of;
+
+use libsql_wal::io::StdIO;
+use libsql_wal::replication::injector::Injector;
+use libsql_wal::segment::Frame as WalFrame;
+use zerocopy::{AsBytes, FromZeroes};
+
+use crate::frame::{Frame, FrameNo};
+
+use super::error::{Error, Result};
+
+pub struct LibsqlInjector {
+    injector: Injector<StdIO>,
+}
+
+impl super::Injector for LibsqlInjector {
+    async fn inject_frame(&mut self, frame: Frame) -> Result<Option<FrameNo>> {
+        // this is a bit annoying be we want to read the frame, and it has to be aligned, so we
+        // must copy it...
+        // FIXME: optimize this.
+        let mut wal_frame = WalFrame::new_box_zeroed();
+        if frame.bytes().len() != size_of::<WalFrame>() {
+            todo!("invalid frame");
+        }
+        wal_frame.as_bytes_mut().copy_from_slice(&frame.bytes()[..]);
+        Ok(self
+            .injector
+            .insert_frame(wal_frame)
+            .await
+            .map_err(|e| Error::FatalInjectError(e.into()))?)
+    }
+
+    async fn rollback(&mut self) {
+        self.injector.rollback();
+    }
+
+    async fn flush(&mut self) -> Result<Option<FrameNo>> {
+        self.injector
+            .flush(None)
+            .await
+            .map_err(|e| Error::FatalInjectError(e.into()))?;
+        Ok(None)
+    }
+}

libsql-replication/src/injector/libsql_injector.rs

@@ -1,6 +1,7 @@
 use std::future::Future;
 
 pub use sqlite_injector::SqliteInjector;
+pub use libsql_injector::LibsqlInjector;
 
 use crate::frame::{Frame, FrameNo};
 
@@ -9,6 +10,7 @@ pub use error::Error;
 
 mod error;
 mod sqlite_injector;
+mod libsql_injector;
 
 pub trait Injector {
     /// Inject a singular frame.

libsql-replication/src/injector/mod.rs

@@ -192,8 +192,8 @@ impl SqliteInjectorInner {
         match stmt.execute(()).and_then(|_| connection.cache_flush()) {
             Ok(_) => panic!("replication hook was not called"),
             Err(e) => {
-                if let Some(e) = e.sqlite_error() {
-                    if e.extended_code == LIBSQL_INJECT_OK {
+                if let Some(err) = e.sqlite_error() {
+                    if err.extended_code == LIBSQL_INJECT_OK {
                         // refresh schema
                         connection.pragma_update(None, "writable_schema", "reset")?;
                         let mut rollback = connection.prepare_cached("ROLLBACK")?;
@@ -203,16 +203,16 @@ impl SqliteInjectorInner {
                         let commit_frame_no = self.biggest_uncommitted_seen;
                         self.biggest_uncommitted_seen = 0;
                         return Ok(Some(commit_frame_no));
-                    } else if e.extended_code == LIBSQL_INJECT_OK_TXN {
+                    } else if err.extended_code == LIBSQL_INJECT_OK_TXN {
                         self.is_txn = true;
                         assert!(self.buffer.lock().is_empty());
                         return Ok(None);
-                    } else if e.extended_code == LIBSQL_INJECT_FATAL {
-                        return Err(Error::FatalInjectError);
+                    } else if err.extended_code == LIBSQL_INJECT_FATAL {
+                        return Err(Error::FatalInjectError(e.into()));
                     }
                 }
 
-                Err(Error::FatalInjectError)
+                Err(Error::FatalInjectError(e.into()))
             }
         }
     }