fix: let replicas keep up with the primary under write load

ARCHITECTURE.md

@@ -231,10 +231,10 @@ replication is a primary-to-replica stream built on top of the same persistence
 
 1. a replica connects and performs a handshake
 2. the primary sends one in-memory snapshot per shard plus the current offset
-3. the replica loads those snapshots
-4. the primary streams incremental `AofRecord` updates
+3. the replica loads those snapshots and acknowledges their offsets
+4. the primary streams incremental `AofRecord` updates, and the replica acknowledges them as it applies them
 
-successful mutations inside a shard emit `ReplicationEvent`s with monotonically increasing per-shard offsets. the replication server uses those offsets for streaming and for the `WAIT` command's acknowledgement tracking.
+each record a shard writes gets the next per-shard offset. a shard collects its records during a burst of requests and sends them as one `ReplicationEvent` on a broadcast channel, so the shards lock the channel once per burst. for each replica, the primary moves those events into a queue of its own, up to 256 MiB. records written during a full sync wait there until the replica has loaded it. a replica that falls further behind is told to resync and starts over with a full sync. the replica applies records in batches, one message per shard. the acknowledged offsets drive the `WAIT` command, which counts a replica only once it has loaded its full sync.
 
 ### slot migration
 

CHANGELOG.md

@@ -11,9 +11,15 @@
 - raft and the `openraft` dependency. raft only ever ran on the bootstrap node, and gossip with config epochs already decided slot ownership and failover, as in Redis Cluster. `raft-port-offset` and `--cluster-raft-port-offset` are still accepted and ignored, so existing configs keep working (#443)
 
 ### performance
+- replication keeps up with the primary under heavy writes. on a 32-core machine a replica loads a full sync of 1.26M keys in about 1 second instead of 13.5, and stays in step with 1.1M writes a second. a replica that fell behind, as every one did under load, was resynced from scratch in a loop and served a partial dataset each time (#455):
+  - the primary queues each replica's records itself, up to 256 MiB, so writes during a full sync no longer overrun the 65,536-event broadcast channel. a replica further behind gets a resync at once instead of first receiving the stale backlog
+  - shards send their records once per burst instead of once per record. every send locks the broadcast channel, which held the primary to about 600k writes a second with a replica attached
+  - the replica applies records and snapshot keys in batches, one message per shard, instead of waiting for each one
 - cluster mode is about as fast as standalone without pipelining: about 860k requests a second against 320k on a 32-core machine. with a pipeline of 16 it serves about 2.3M against 320k, 15% below standalone. every command locked the migration state, an async mutex shared by every connection, and read the cluster state up to three times. it now reads each once, and the migration state takes a shared lock (#453)
 
 ### fixes
+- WAIT counts a replica only once it has loaded its full sync. it counted the replica as soon as the primary had sent the snapshot, while the replica was still loading it, or had just flushed its data to start over (#455)
+- `INFO replication` on a primary counts the replicas streaming from it in `connected_replicas`. it counted the replicas gossip had reported, which could read 0 while one streamed. a new `sync_full` field counts the full syncs sent (#455)
 - failover safety (#452):
   - an election needs a majority of all primaries, the failed one included, as in Redis Cluster. it counted only the other primaries, so with two primaries a replica cut off with one of them won with a single vote while its primary kept serving
   - a primary votes for one replica per failed primary for two failure timeouts, so two replicas of the same primary can't both win. a replica stops its election once another node owns the failed primary's slots

crates/ember-core/src/shard/aof.rs

@@ -886,24 +886,60 @@ impl AofLog {
     }
 }
 
-/// Broadcasts a single replication event, used by blocking pop operations
-/// that bypass the normal process_message path.
-pub(super) fn broadcast_replication(
-    record: AofRecord,
-    replication_tx: &Option<broadcast::Sender<ReplicationEvent>>,
-    replication_offset: &AtomicU64,
+/// Collects a shard's replication records during a burst and sends them
+/// to the replicas as one event at its end.
+///
+/// Every shard sends into one broadcast channel, and each send locks it,
+/// so one send per record caps the primary's write rate.
+pub(super) struct ReplicationLog {
+    tx: broadcast::Sender<ReplicationEvent>,
+    /// Offset of the last record taken. Shared with the `ShardHandle`, so
+    /// WAIT can read it as soon as the write is applied.
+    offset: Arc<AtomicU64>,
+    pending: Vec<AofRecord>,
     shard_id: u16,
-) {
-    if let Some(ref tx) = *replication_tx {
-        let offset = replication_offset.fetch_add(1, Ordering::Release) + 1;
-        if tx
-            .send(ReplicationEvent {
-                shard_id,
-                offset,
-                record,
-            })
-            .is_err()
-        {
+}
+
+impl ReplicationLog {
+    pub(super) fn new(
+        tx: broadcast::Sender<ReplicationEvent>,
+        offset: Arc<AtomicU64>,
+        shard_id: u16,
+    ) -> Self {
+        Self {
+            tx,
+            offset,
+            pending: Vec::new(),
+            shard_id,
+        }
+    }
+
+    /// The offset of the last record taken.
+    pub(super) fn offset(&self) -> u64 {
+        self.offset.load(Ordering::Acquire)
+    }
+
+    /// Takes records applied by a write, numbering each with the next
+    /// offset.
+    pub(super) fn push(&mut self, records: impl IntoIterator<Item = AofRecord>) {
+        for record in records {
+            self.offset.fetch_add(1, Ordering::Release);
+            self.pending.push(record);
+        }
+    }
+
+    /// Sends the records taken since the last call as one event.
+    pub(super) fn flush(&mut self) {
+        if self.pending.is_empty() {
+            return;
+        }
+        let records: Arc<[AofRecord]> = std::mem::take(&mut self.pending).into();
+        let event = ReplicationEvent {
+            shard_id: self.shard_id,
+            offset: self.offset() + 1 - records.len() as u64,
+            records,
+        };
+        if self.tx.send(event).is_err() {
             // no replicas are currently connected — normal during startup
             // or after a replica disconnects, but tracked so operators
             // can alert on unexpected drops in a replicated deployment.

crates/ember-core/src/shard/blocking.rs

@@ -115,12 +115,9 @@ fn pop_for(
     if let Some(ref mut aof) = *ctx.aof {
         aof.append(std::slice::from_ref(&record));
     }
-    aof::broadcast_replication(
-        record,
-        ctx.replication_tx,
-        ctx.replication_offset,
-        ctx.shard_id,
-    );
+    if let Some(ref mut replication) = *ctx.replication {
+        replication.push([record]);
+    }
     true
 }
 

crates/ember-core/src/shard/mod.rs

@@ -78,19 +78,20 @@ use crate::types::sorted_set::{ScoreBound, ZAddFlags};
 use crate::types::Value;
 use ember_protocol::command::{BitOpKind, BitRange};
 
-/// A mutation event broadcast to replication subscribers.
+/// Mutations broadcast to replication subscribers, one event per burst of
+/// requests a shard handles.
 ///
-/// 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.
+/// Each record has a per-shard offset that increases by one per record.
+/// The first record's offset is `offset`, the next one `offset + 1`, and
+/// so on.
 #[derive(Debug, Clone)]
 pub struct ReplicationEvent {
     /// The shard that produced this event.
     pub shard_id: u16,
-    /// Monotonically increasing per-shard offset.
+    /// Offset of the first record.
     pub offset: u64,
-    /// The mutation record, ready to replay on a replica.
-    pub record: AofRecord,
+    /// The mutation records, in order, ready to replay on a replica.
+    pub records: Arc<[AofRecord]>,
 }
 
 /// Optional persistence configuration for a shard.
@@ -312,6 +313,8 @@ async fn run_shard(prepared: PreparedShard) {
         schema_registry,
     } = prepared;
     let shard_id = config.shard_id;
+    let mut replication =
+        replication_tx.map(|tx| aof::ReplicationLog::new(tx, replication_offset, shard_id));
     // tokio's interval panics on zero, so hold both to at least 1ms
     let expiry_interval = config.expiry_interval.max(Duration::from_millis(1));
     let fsync_interval = config.fsync_interval.max(Duration::from_millis(1));
@@ -432,8 +435,7 @@ async fn run_shard(prepared: PreparedShard) {
                             persistence: &persistence,
                             drop_handle: &drop_handle,
                             shard_id,
-                            replication_tx: &replication_tx,
-                            replication_offset: &replication_offset,
+                            replication: &mut replication,
                             lpop_waiters: &mut lpop_waiters,
                             rpop_waiters: &mut rpop_waiters,
                             #[cfg(feature = "protobuf")]
@@ -444,6 +446,9 @@ async fn run_shard(prepared: PreparedShard) {
                         if let Some(ref mut aof) = aof {
                             aof.flush();
                         }
+                        if let Some(ref mut replication) = replication {
+                            replication.flush();
+                        }
                         for (reply, response) in held_replies.drain(..) {
                             reply.send(response);
                         }
@@ -487,10 +492,8 @@ struct ProcessCtx<'a> {
     persistence: &'a Option<ShardPersistenceConfig>,
     drop_handle: &'a Option<DropHandle>,
     shard_id: u16,
-    replication_tx: &'a Option<broadcast::Sender<ReplicationEvent>>,
-    /// Offset of the last replication event sent. Shared with the
-    /// `ShardHandle` so WAIT can read it.
-    replication_offset: &'a AtomicU64,
+    /// Where replication records wait for the end of the burst.
+    replication: &'a mut Option<aof::ReplicationLog>,
     /// Waiters for BLPOP — keyed by list name, FIFO order.
     lpop_waiters: &'a mut HashMap<String, VecDeque<mpsc::Sender<(String, Bytes)>>>,
     /// Waiters for BRPOP — keyed by list name, FIFO order.
@@ -666,16 +669,8 @@ fn process_single(mut request: ShardRequest, reply: ReplySender, ctx: &mut Proce
     }
 
     // broadcast mutation events to replication subscribers
-    if let Some(ref tx) = *ctx.replication_tx {
-        for record in records {
-            let offset = ctx.replication_offset.fetch_add(1, Ordering::Release) + 1;
-            // ignore send errors — no subscribers or lagged consumers
-            let _ = tx.send(ReplicationEvent {
-                shard_id,
-                offset,
-                record,
-            });
-        }
+    if let Some(ref mut replication) = *ctx.replication {
+        replication.push(records);
     }
 
     if let Some(key) = pushed_key {
@@ -685,9 +680,10 @@ fn process_single(mut request: ShardRequest, reply: ReplySender, ctx: &mut Proce
     // handle special requests that need access to persistence state
     match request_kind {
         RequestKind::SerializeSnapshot => {
-            // the snapshot covers every event up to the current offset,
-            // since this task sends all of them and is busy doing this now
-            let offset = ctx.replication_offset.load(Ordering::Acquire);
+            // the snapshot covers every record up to the current offset,
+            // including those not yet sent: they go out later with offsets
+            // the replica knows to skip
+            let offset = ctx.replication.as_ref().map_or(0, |r| r.offset());
             let resp = persistence::handle_serialize_snapshot(ctx.keyspace, shard_id, offset);
             ctx.respond(reply, resp);
             return;

crates/ember-server/src/cluster/replication.rs

@@ -11,8 +11,6 @@ pub struct ReplicationInfo {
     pub role: NodeRole,
     /// Address of the primary this node replicates from (replica only).
     pub primary_addr: Option<std::net::SocketAddr>,
-    /// Number of connected replicas (primary only).
-    pub replica_count: usize,
 }
 
 impl ClusterCoordinator {
@@ -136,8 +134,6 @@ impl ClusterCoordinator {
     }
 
     /// Returns replication status for the `INFO replication` section.
-    ///
-    /// Returns `(role, Option<primary_addr>)`.
     pub async fn replication_info(&self) -> ReplicationInfo {
         let state = self.state.read().await;
         let local = state.nodes.get(&self.local_id);
@@ -150,15 +146,6 @@ impl ClusterCoordinator {
         } else {
             None
         };
-        let replica_count = if role == NodeRole::Primary {
-            local.map(|n| n.replicas.len()).unwrap_or(0)
-        } else {
-            0
-        };
-        ReplicationInfo {
-            role,
-            primary_addr,
-            replica_count,
-        }
+        ReplicationInfo { role, primary_addr }
     }
 }

crates/ember-server/src/connection/exec/server.rs

@@ -379,7 +379,14 @@ async fn render_info(
             match info.role {
                 NodeRole::Primary => {
                     out.push_str("role:primary\r\n");
-                    out.push_str(&format!("connected_replicas:{}\r\n", info.replica_count));
+                    out.push_str(&format!(
+                        "connected_replicas:{}\r\n",
+                        ctx.replica_tracker.connected_count()
+                    ));
+                    out.push_str(&format!(
+                        "sync_full:{}\r\n",
+                        ctx.replica_tracker.full_syncs()
+                    ));
                 }
                 NodeRole::Replica => {
                     out.push_str("role:replica\r\n");