feat(replication): primary → replica data sync stream

crates/ember-core/src/engine.rs

@@ -4,10 +4,12 @@
 //! of the key. Each shard is an independent tokio task — no locks on
 //! the hot path.
 
+use tokio::sync::broadcast;
+
 use crate::dropper::DropHandle;
 use crate::error::ShardError;
 use crate::keyspace::ShardConfig;
-use crate::shard::{self, ShardHandle, ShardPersistenceConfig, ShardRequest, ShardResponse};
+use crate::shard::{self, ReplicationEvent, ShardHandle, ShardPersistenceConfig, ShardRequest, ShardResponse};
 
 /// Channel buffer size per shard. 256 is large enough to absorb
 /// bursts without putting meaningful back-pressure on connections.
@@ -21,6 +23,12 @@ pub struct EngineConfig {
     /// Optional persistence configuration. When set, each shard gets
     /// its own AOF and snapshot files under this directory.
     pub persistence: Option<ShardPersistenceConfig>,
+    /// Optional broadcast sender for replication events.
+    ///
+    /// When set, every successful mutation is published as a
+    /// [`ReplicationEvent`] so replication clients can stream it to
+    /// replicas.
+    pub replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
     /// Optional schema registry for protobuf value validation.
     /// When set, enables PROTO.* commands.
     #[cfg(feature = "protobuf")]
@@ -35,6 +43,7 @@ pub struct EngineConfig {
 #[derive(Debug, Clone)]
 pub struct Engine {
     shards: Vec<ShardHandle>,
+    replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
     #[cfg(feature = "protobuf")]
     schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 }
@@ -72,6 +81,7 @@ impl Engine {
                     shard_config,
                     config.persistence.clone(),
                     Some(drop_handle.clone()),
+                    config.replication_tx.clone(),
                     #[cfg(feature = "protobuf")]
                     config.schema_registry.clone(),
                 )
@@ -80,6 +90,7 @@ impl Engine {
 
         Self {
             shards,
+            replication_tx: config.replication_tx,
             #[cfg(feature = "protobuf")]
             schema_registry: config.schema_registry,
         }
@@ -112,6 +123,15 @@ impl Engine {
         self.shards.len()
     }
 
+    /// Creates a new broadcast receiver for replication events.
+    ///
+    /// Returns `None` if no replication channel was configured. Each
+    /// caller gets an independent receiver starting from the current
+    /// broadcast position — not from the beginning of the stream.
+    pub fn subscribe_replication(&self) -> Option<broadcast::Receiver<ReplicationEvent>> {
+        self.replication_tx.as_ref().map(|tx| tx.subscribe())
+    }
+
     /// Sends a request to a specific shard by index.
     ///
     /// Used by SCAN to iterate through shards sequentially.

crates/ember-core/src/lib.rs

@@ -27,5 +27,5 @@ pub use keyspace::{
 };
 #[cfg(feature = "vector")]
 pub use keyspace::{VAddResult, VectorWriteError};
-pub use shard::{ShardPersistenceConfig, ShardRequest, ShardResponse};
+pub use shard::{ReplicationEvent, ShardHandle, ShardPersistenceConfig, ShardRequest, ShardResponse};
 pub use types::Value;

crates/ember-core/src/shard.rs

@@ -12,7 +12,7 @@ use bytes::Bytes;
 use ember_persistence::aof::{AofRecord, AofWriter, FsyncPolicy};
 use ember_persistence::recovery::{self, RecoveredValue};
 use ember_persistence::snapshot::{self, SnapEntry, SnapValue, SnapshotWriter};
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{broadcast, mpsc, oneshot};
 use tracing::{info, warn};
 
 use crate::dropper::DropHandle;
@@ -32,6 +32,21 @@ const EXPIRY_TICK: Duration = Duration::from_millis(100);
 /// How often to fsync when using the `EverySec` policy.
 const FSYNC_INTERVAL: Duration = Duration::from_secs(1);
 
+/// A mutation event broadcast to replication subscribers.
+///
+/// Published after every successful mutation on the hot path. The
+/// `offset` is per-shard and monotonically increasing — replicas use it
+/// to detect gaps and trigger re-sync when they fall behind.
+#[derive(Debug, Clone)]
+pub struct ReplicationEvent {
+    /// The shard that produced this event.
+    pub shard_id: u16,
+    /// Monotonically increasing per-shard offset.
+    pub offset: u64,
+    /// The mutation record, ready to replay on a replica.
+    pub record: AofRecord,
+}
+
 /// Optional persistence configuration for a shard.
 #[derive(Debug, Clone)]
 pub struct ShardPersistenceConfig {
@@ -239,6 +254,11 @@ pub enum ShardRequest {
     Stats,
     /// Triggers a snapshot write.
     Snapshot,
+    /// Serializes the current shard state to bytes (in-memory snapshot).
+    ///
+    /// Used by the replication server to capture a consistent shard
+    /// snapshot for transmission to a new replica without filesystem I/O.
+    SerializeSnapshot,
     /// Triggers an AOF rewrite (snapshot + truncate AOF).
     RewriteAof,
     /// Clears all keys from the keyspace.
@@ -427,6 +447,8 @@ pub enum ShardResponse {
     StringArray(Vec<String>),
     /// Serialized key dump with remaining TTL (for MIGRATE/DUMP).
     KeyDump { data: Vec<u8>, ttl_ms: i64 },
+    /// In-memory snapshot of the full shard state (for replication).
+    SnapshotData { shard_id: u16, data: Vec<u8> },
     /// HMGET result: array of optional values.
     OptionalArray(Vec<Option<Bytes>>),
     /// VADD result: element, vector, and whether it was newly added.
@@ -523,6 +545,7 @@ pub fn spawn_shard(
     config: ShardConfig,
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
+    replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
     #[cfg(feature = "protobuf")] schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 ) -> ShardHandle {
     let (tx, rx) = mpsc::channel(buffer);
@@ -531,6 +554,7 @@ pub fn spawn_shard(
         config,
         persistence,
         drop_handle,
+        replication_tx,
         #[cfg(feature = "protobuf")]
         schema_registry,
     ));
@@ -544,6 +568,7 @@ async fn run_shard(
     config: ShardConfig,
     persistence: Option<ShardPersistenceConfig>,
     drop_handle: Option<DropHandle>,
+    replication_tx: Option<broadcast::Sender<ReplicationEvent>>,
     #[cfg(feature = "protobuf")] schema_registry: Option<crate::schema::SharedSchemaRegistry>,
 ) {
     let shard_id = config.shard_id;
@@ -670,6 +695,9 @@ async fn run_shard(
         .map(|p| p.fsync_policy)
         .unwrap_or(FsyncPolicy::No);
 
+    // monotonically increasing per-shard replication offset
+    let mut replication_offset: u64 = 0;
+
     // -- tickers --
     let mut expiry_tick = tokio::time::interval(EXPIRY_TICK);
     expiry_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -690,6 +718,8 @@ async fn run_shard(
                             &persistence,
                             &drop_handle,
                             shard_id,
+                            &replication_tx,
+                            &mut replication_offset,
                             #[cfg(feature = "protobuf")]
                             &schema_registry,
                         );
@@ -707,6 +737,8 @@ async fn run_shard(
                                 &persistence,
                                 &drop_handle,
                                 shard_id,
+                                &replication_tx,
+                                &mut replication_offset,
                                 #[cfg(feature = "protobuf")]
                                 &schema_registry,
                             );
@@ -748,6 +780,8 @@ fn process_message(
     persistence: &Option<ShardPersistenceConfig>,
     drop_handle: &Option<DropHandle>,
     shard_id: u16,
+    replication_tx: &Option<broadcast::Sender<ReplicationEvent>>,
+    replication_offset: &mut u64,
     #[cfg(feature = "protobuf")] schema_registry: &Option<crate::schema::SharedSchemaRegistry>,
 ) {
     let request_kind = describe_request(&msg.request);
@@ -758,9 +792,11 @@ fn process_message(
         schema_registry,
     );
 
+    // collect mutation records once; used for both AOF and replication
+    let records = to_aof_records(&msg.request, &response);
+
     // write AOF records for successful mutations
     if let Some(ref mut writer) = aof_writer {
-        let records = to_aof_records(&msg.request, &response);
         for record in &records {
             if let Err(e) = writer.write_record(record) {
                 warn!(shard_id, "aof write failed: {e}");
@@ -773,13 +809,31 @@ fn process_message(
         }
     }
 
+    // broadcast mutation events to replication subscribers
+    if let Some(ref tx) = replication_tx {
+        for record in records {
+            *replication_offset += 1;
+            // ignore send errors — no subscribers or lagged consumers
+            let _ = tx.send(ReplicationEvent {
+                shard_id,
+                offset: *replication_offset,
+                record,
+            });
+        }
+    }
+
     // handle special requests that need access to persistence state
     match request_kind {
         RequestKind::Snapshot => {
             let resp = handle_snapshot(keyspace, persistence, shard_id);
             let _ = msg.reply.send(resp);
             return;
         }
+        RequestKind::SerializeSnapshot => {
+            let resp = handle_serialize_snapshot(keyspace, shard_id);
+            let _ = msg.reply.send(resp);
+            return;
+        }
         RequestKind::RewriteAof => {
             let resp = handle_rewrite(
                 keyspace,
@@ -810,6 +864,7 @@ fn process_message(
 /// handling after dispatch without borrowing the request again.
 enum RequestKind {
     Snapshot,
+    SerializeSnapshot,
     RewriteAof,
     FlushDbAsync,
     Other,
@@ -818,6 +873,7 @@ enum RequestKind {
 fn describe_request(req: &ShardRequest) -> RequestKind {
     match req {
         ShardRequest::Snapshot => RequestKind::Snapshot,
+        ShardRequest::SerializeSnapshot => RequestKind::SerializeSnapshot,
         ShardRequest::RewriteAof => RequestKind::RewriteAof,
         ShardRequest::FlushDbAsync => RequestKind::FlushDbAsync,
         _ => RequestKind::Other,
@@ -1269,10 +1325,11 @@ fn dispatch(
                 })
             })
         }
-        // snapshot/rewrite/flush_async are handled in the main loop, not here
-        ShardRequest::Snapshot | ShardRequest::RewriteAof | ShardRequest::FlushDbAsync => {
-            ShardResponse::Ok
-        }
+        // these requests are intercepted in process_message, not handled here
+        ShardRequest::Snapshot
+        | ShardRequest::SerializeSnapshot
+        | ShardRequest::RewriteAof
+        | ShardRequest::FlushDbAsync => ShardResponse::Ok,
     }
 }
 
@@ -1611,6 +1668,30 @@ fn handle_snapshot(
     }
 }
 
+/// Serializes the current shard state to bytes without filesystem I/O.
+///
+/// Used by the replication server to capture a snapshot for transmission
+/// to a new replica. The format matches the file-based snapshot and can
+/// be loaded via [`ember_persistence::snapshot::read_snapshot_from_bytes`].
+fn handle_serialize_snapshot(keyspace: &Keyspace, shard_id: u16) -> ShardResponse {
+    let entries: Vec<SnapEntry> = keyspace
+        .iter_entries()
+        .map(|(key, value, expire_ms)| SnapEntry {
+            key: key.to_owned(),
+            value: value_to_snap(value),
+            expire_ms,
+        })
+        .collect();
+
+    match snapshot::write_snapshot_bytes(shard_id, &entries) {
+        Ok(data) => ShardResponse::SnapshotData { shard_id, data },
+        Err(e) => {
+            warn!(shard_id, "snapshot serialization failed: {e}");
+            ShardResponse::Err(format!("snapshot failed: {e}"))
+        }
+    }
+}
+
 /// Writes a snapshot and then truncates the AOF.
 ///
 /// When protobuf is enabled, re-persists all registered schemas to the
@@ -1892,6 +1973,7 @@ mod tests {
             ShardConfig::default(),
             None,
             None,
+            None,
             #[cfg(feature = "protobuf")]
             None,
         );
@@ -1929,6 +2011,7 @@ mod tests {
             ShardConfig::default(),
             None,
             None,
+            None,
             #[cfg(feature = "protobuf")]
             None,
         );
@@ -1960,6 +2043,7 @@ mod tests {
             ShardConfig::default(),
             None,
             None,
+            None,
             #[cfg(feature = "protobuf")]
             None,
         );
@@ -2033,6 +2117,7 @@ mod tests {
                 config.clone(),
                 Some(pcfg.clone()),
                 None,
+                None,
                 #[cfg(feature = "protobuf")]
                 None,
             );
@@ -2081,6 +2166,7 @@ mod tests {
                 config,
                 Some(pcfg),
                 None,
+                None,
                 #[cfg(feature = "protobuf")]
                 None,
             );

crates/ember-persistence/src/aof.rs

@@ -333,7 +333,7 @@ impl AofRecord {
     }
 
     /// Serializes this record into a byte vector (tag + payload, no CRC).
-    fn to_bytes(&self) -> Result<Vec<u8>, FormatError> {
+    pub fn to_bytes(&self) -> Result<Vec<u8>, FormatError> {
         let mut buf = Vec::with_capacity(self.estimated_size());
         format::write_u8(&mut buf, self.tag())?;
 
@@ -489,7 +489,7 @@ impl AofRecord {
     }
 
     /// Deserializes a record from a byte slice (tag + payload, no CRC).
-    fn from_bytes(data: &[u8]) -> Result<Self, FormatError> {
+    pub fn from_bytes(data: &[u8]) -> Result<Self, FormatError> {
         let mut cursor = io::Cursor::new(data);
         let tag = format::read_u8(&mut cursor)?;
         match tag {