feat: added replcator crate with basic protocol impl

Cargo.toml

@@ -19,6 +19,7 @@ members = [
   "magicblock-ledger",
   "magicblock-metrics",
   "magicblock-processor",
+  "magicblock-replicator",
   "magicblock-rpc-client",
   "magicblock-table-mania",
   "magicblock-task-scheduler",
@@ -54,6 +55,7 @@ assert_matches = "1.5.0"
 async-trait = "0.1.77"
 base64 = "0.21.7"
 bincode = "1.3.3"
+bytes = "1.0"
 borsh = { version = "1.5.1", features = ["derive", "unstable__schema"] }
 bs58 = "0.5.1"
 byteorder = "1.5.0"

magicblock-replicator/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "magicblock-replicator"
+version.workspace = true
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+license.workspace = true
+edition.workspace = true
+
+[dependencies]
+bincode = { workspace = true }
+bytes = { workspace = true }
+futures = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["net", "rt", "macros"] }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true, features = ["codec"] }
+serde = { workspace = true, features = ["derive"] }
+solana-hash = { workspace = true, features = ["serde"] }
+solana-keypair = { workspace = true }
+solana-pubkey = { workspace = true, features = ["serde"] }
+solana-signature = { workspace = true, features = ["serde"] }
+solana-signer = { workspace = true }
+solana-transaction = { workspace = true, features = ["serde"] }

magicblock-replicator/src/connection.rs

@@ -0,0 +1,104 @@
+//! Codec and stream types for length-prefixed bincode framing.
+
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::{BufMut, BytesMut};
+use futures::{SinkExt, Stream, StreamExt};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_util::codec::{
+    Encoder, FramedRead, FramedWrite, LengthDelimitedCodec,
+};
+
+use crate::{
+    error::{Error, Result},
+    proto::Message,
+};
+
+/// Encodes `Message` with 4-byte LE length prefix.
+pub(crate) struct MessageEncoder;
+
+pub(crate) type InputStream<IO> = FramedRead<IO, LengthDelimitedCodec>;
+pub(crate) type OutputStream<IO> = FramedWrite<IO, MessageEncoder>;
+
+const MAX_FRAME_SIZE: usize = 64 * 1024 * 1024;
+
+impl Encoder<&Message> for MessageEncoder {
+    type Error = Error;
+
+    fn encode(&mut self, msg: &Message, dst: &mut BytesMut) -> Result<()> {
+        let start = dst.len();
+        dst.put_u32_le(0);
+        bincode::serialize_into(dst.writer(), msg)?;
+        let len = dst.len() - start - 4;
+        if len > MAX_FRAME_SIZE {
+            dst.truncate(start);
+            return Err(Box::new(bincode::ErrorKind::SizeLimit))?;
+        }
+        dst[start..start + 4].copy_from_slice(&(len as u32).to_le_bytes());
+        Ok(())
+    }
+}
+
+/// Receives messages from an async stream (max frame: 64MB).
+pub struct Receiver<IO> {
+    inner: InputStream<IO>,
+}
+
+impl<IO: AsyncRead + Unpin> Receiver<IO> {
+    pub fn new(io: IO) -> Self {
+        let inner = LengthDelimitedCodec::builder()
+            .little_endian()
+            .max_frame_length(MAX_FRAME_SIZE)
+            .length_field_type::<u32>()
+            .new_read(io);
+        Self { inner }
+    }
+
+    pub async fn recv(&mut self) -> Result<Message> {
+        let frame =
+            self.inner.next().await.ok_or(Error::ConnectionClosed)??;
+        bincode::deserialize(&frame).map_err(Into::into)
+    }
+}
+
+impl<IO: AsyncRead + Unpin> Stream for Receiver<IO> {
+    type Item = Result<Message>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        match self.inner.poll_next_unpin(cx) {
+            Poll::Ready(Some(Ok(item))) => {
+                let result = bincode::deserialize(&item).map_err(Into::into);
+                Poll::Ready(Some(result))
+            }
+            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err.into()))),
+            Poll::Ready(None) => {
+                Poll::Ready(Some(Err(Error::ConnectionClosed)))
+            }
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+/// Sends messages to an async stream.
+pub struct Sender<IO> {
+    inner: OutputStream<IO>,
+}
+
+impl<IO: AsyncWrite + Unpin> Sender<IO> {
+    pub fn new(io: IO) -> Self {
+        Self {
+            inner: FramedWrite::new(io, MessageEncoder),
+        }
+    }
+
+    pub async fn send(&mut self, msg: &Message) -> Result<()> {
+        self.inner.send(msg).await?;
+        Ok(())
+    }
+}

magicblock-replicator/src/error.rs

@@ -0,0 +1,14 @@
+//! Error types for the replication protocol.
+
+/// Replication operation errors.
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("connection closed")]
+    ConnectionClosed,
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("serialization error: {0}")]
+    SerDe(#[from] bincode::Error),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

magicblock-replicator/src/lib.rs

@@ -0,0 +1,14 @@
+//! State replication protocol for streaming transactions from primary to standby nodes.
+//!
+//! Messages are length-prefixed (4B LE) + bincode payload.
+
+pub mod connection;
+pub mod error;
+pub mod proto;
+pub mod tcp;
+
+#[cfg(test)]
+mod tests;
+
+pub use error::{Error, Result};
+pub use proto::{Message, PROTOCOL_VERSION};

magicblock-replicator/src/proto.rs

@@ -0,0 +1,146 @@
+//! Protocol message types for replication.
+//!
+//! Wire format: 4-byte LE length prefix + bincode payload.
+//! Bincode encodes enum variant index as implicit type tag.
+
+use serde::{Deserialize, Serialize};
+use solana_hash::Hash;
+use solana_keypair::Keypair;
+use solana_pubkey::Pubkey;
+use solana_signature::Signature;
+use solana_signer::Signer;
+use solana_transaction::versioned::VersionedTransaction;
+
+use crate::error::Result;
+
+pub type Slot = u64;
+pub type TxIndex = u32;
+
+pub const PROTOCOL_VERSION: u32 = 1;
+
+/// Top-level replication message.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub enum Message {
+    HandshakeReq(HandshakeRequest),
+    HandshakeResp(HandshakeResponse),
+    Transaction(Transaction),
+    Block(Block),
+    SuperBlock(SuperBlock),
+    Failover(FailoverSignal),
+}
+
+/// Client -> Server: initiate replication session.
+/// Authenticated via Ed25519 signature over `start_slot`.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct HandshakeRequest {
+    pub version: u32,
+    pub start_slot: Slot,
+    pub identity: Pubkey,
+    signature: Signature,
+}
+
+/// Server -> Client: accept or reject session.
+/// Signed over `slot` (success) or error message (failure).
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct HandshakeResponse {
+    pub result: std::result::Result<Slot, String>,
+    pub identity: Pubkey,
+    signature: Signature,
+}
+
+/// Slot boundary marker with blockhash.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct Block {
+    pub slot: Slot,
+    pub hash: Hash,
+}
+
+/// Transaction with slot and ordinal position.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct Transaction {
+    pub slot: Slot,
+    pub index: TxIndex,
+    /// Bincode-encoded `VersionedTransaction`.
+    pub payload: Vec<u8>,
+}
+
+/// Periodic checkpoint for state verification.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct SuperBlock {
+    pub blocks: u64,
+    pub transactions: u64,
+    pub checksum: u64,
+}
+
+/// Primary -> Standby: signal controlled failover.
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub struct FailoverSignal {
+    pub slot: Slot,
+    signature: Signature,
+}
+
+impl HandshakeRequest {
+    pub fn new(start_slot: Slot, keypair: &Keypair) -> Self {
+        Self {
+            version: PROTOCOL_VERSION,
+            start_slot,
+            identity: keypair.pubkey(),
+            signature: keypair.sign_message(&start_slot.to_le_bytes()),
+        }
+    }
+
+    /// Verifies signature matches claimed identity.
+    pub fn verify(&self) -> bool {
+        self.signature
+            .verify(self.identity.as_array(), &self.start_slot.to_le_bytes())
+    }
+}
+
+impl HandshakeResponse {
+    pub fn new(result: Result<Slot>, keypair: &Keypair) -> Self {
+        let result = result.map_err(|e| e.to_string());
+        let signature = match &result {
+            Ok(slot) => keypair.sign_message(&slot.to_le_bytes()),
+            Err(err) => keypair.sign_message(err.as_bytes()),
+        };
+        Self {
+            result,
+            identity: keypair.pubkey(),
+            signature,
+        }
+    }
+
+    /// Verifies signature matches server identity.
+    pub fn verify(&self) -> bool {
+        match &self.result {
+            Ok(slot) => self
+                .signature
+                .verify(self.identity.as_array(), &slot.to_le_bytes()),
+            Err(err) => self
+                .signature
+                .verify(self.identity.as_array(), err.as_bytes()),
+        }
+    }
+}
+
+impl Transaction {
+    /// Deserializes the inner transaction.
+    pub fn decode(&self) -> bincode::Result<VersionedTransaction> {
+        bincode::deserialize(&self.payload)
+    }
+}
+
+impl FailoverSignal {
+    pub fn new(slot: Slot, keypair: &Keypair) -> Self {
+        Self {
+            slot,
+            signature: keypair.sign_message(&slot.to_le_bytes()),
+        }
+    }
+
+    /// Verifies signal against expected identity.
+    pub fn verify(&self, identity: Pubkey) -> bool {
+        self.signature
+            .verify(identity.as_array(), &self.slot.to_le_bytes())
+    }
+}

magicblock-replicator/src/tcp.rs

@@ -0,0 +1,24 @@
+//! TCP transport utilities.
+
+use std::{io, net::SocketAddr};
+
+use tokio::net::{
+    tcp::{OwnedReadHalf, OwnedWriteHalf},
+    TcpStream,
+};
+
+use crate::connection::{Receiver, Sender};
+
+pub type TcpReceiver = Receiver<OwnedReadHalf>;
+pub type TcpSender = Sender<OwnedWriteHalf>;
+
+/// Connects to a primary at `addr`, returning (sender, receiver).
+pub async fn connect(addr: SocketAddr) -> io::Result<(TcpSender, TcpReceiver)> {
+    TcpStream::connect(addr).await.map(split)
+}
+
+/// Splits a TCP stream into sender and receiver halves.
+pub fn split(stream: TcpStream) -> (TcpSender, TcpReceiver) {
+    let (rx, tx) = stream.into_split();
+    (Sender::new(tx), Receiver::new(rx))
+}