feat(message_bus): read replica sockets ahead, bypassing large frames

core/configs/src/server_config/defaults.rs

@@ -296,6 +296,7 @@ impl Default for MessageBusConfig {
         MessageBusConfig {
             max_batch: bus.max_batch as usize,
             max_message_size: bus.max_message_size.parse().unwrap(),
+            replica_read_buffer_size: bus.replica_read_buffer_size.parse().unwrap(),
             peer_queue_capacity: bus.peer_queue_capacity as usize,
             client_queue_capacity: bus.client_queue_capacity as usize,
             reconnect_period: bus.reconnect_period.parse().unwrap(),

core/configs/src/server_config/displays.rs

@@ -96,12 +96,14 @@ impl Display for MessageBusConfig {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "{{ max_batch: {}, max_message_size: {}, peer_queue_capacity: {}, \
-             reconnect_period: {}, close_peer_timeout: {}, close_grace: {}, \
-             handshake_grace: {} }}",
+            "{{ max_batch: {}, max_message_size: {}, replica_read_buffer_size: {}, \
+             peer_queue_capacity: {}, client_queue_capacity: {}, reconnect_period: {}, \
+             close_peer_timeout: {}, close_grace: {}, handshake_grace: {} }}",
             self.max_batch,
             self.max_message_size,
+            self.replica_read_buffer_size,
             self.peer_queue_capacity,
+            self.client_queue_capacity,
             self.reconnect_period,
             self.close_peer_timeout,
             self.close_grace,

core/configs/src/server_config/message_bus.rs

@@ -52,6 +52,7 @@
 //! [`super::server::ServerConfig::load`].
 
 use super::COMPONENT;
+use super::defaults::SERVER_CONFIG;
 use crate::ConfigurationError;
 use configs::ConfigEnv;
 use iggy_common::{IggyByteSize, IggyDuration, MAX_MESSAGE_SIZE_UPPER_BYTES, Validatable};
@@ -70,7 +71,9 @@ use serde_with::{DisplayFromStr, serde_as};
 /// failure until both are reconciled.
 pub const IOV_MAX_LIMIT: usize = 512;
 
-const DEFAULT_CLIENT_QUEUE_CAPACITY: usize = 256;
+/// Floor for a nonzero [`MessageBusConfig::replica_read_buffer_size`].
+/// A buffer below one page costs more reads and copies than no buffer.
+const MIN_REPLICA_READ_BUFFER_BYTES: u64 = 4 * 1024;
 
 /// Tunables for the message bus that ships consensus traffic between
 /// replicas and SDK-client traffic between shards.
@@ -88,6 +91,18 @@ pub struct MessageBusConfig {
     #[config_env(leaf)]
     pub max_message_size: IggyByteSize,
 
+    /// Read-ahead buffer per plaintext replica link. Costs one buffer of
+    /// this size per installed connection. The framing layer decodes
+    /// every complete frame a fill delivered, so a burst costs one read
+    /// instead of one or two per frame. A body at least this size is
+    /// read straight into its frame rather than through the buffer. Zero
+    /// keeps the unbuffered path. No effect under `cluster.tls`, whose
+    /// reader buffers inside compio's `SyncStream` adapter, nor on the
+    /// client plane, which never wraps its read half.
+    #[serde(default = "default_replica_read_buffer_size")]
+    #[config_env(leaf)]
+    pub replica_read_buffer_size: IggyByteSize,
+
     /// Bound on each replica peer's mpsc queue. Writer task drains; the
     /// `send_to_*` path enqueues. Too small drops under burst; too
     /// large delays backpressure signalling.
@@ -145,6 +160,17 @@ impl Validatable<ConfigurationError> for MessageBusConfig {
             );
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
+        // The one key in this section where zero is legal: it selects the
+        // unbuffered read path, so the A/B baseline is a config change
+        // rather than a separate build.
+        let replica_read_buffer = self.replica_read_buffer_size.as_bytes_u64();
+        if replica_read_buffer != 0 && replica_read_buffer < MIN_REPLICA_READ_BUFFER_BYTES {
+            eprintln!(
+                "{COMPONENT} message_bus.replica_read_buffer_size ({replica_read_buffer}) must be \
+                 0 (unbuffered) or at least {MIN_REPLICA_READ_BUFFER_BYTES} bytes"
+            );
+            return Err(ConfigurationError::InvalidConfigurationValue);
+        }
         if self.peer_queue_capacity == 0 {
             eprintln!("{COMPONENT} message_bus.peer_queue_capacity must be > 0");
             return Err(ConfigurationError::InvalidConfigurationValue);
@@ -187,8 +213,19 @@ impl Validatable<ConfigurationError> for MessageBusConfig {
     }
 }
 
-const fn default_client_queue_capacity() -> usize {
-    DEFAULT_CLIENT_QUEUE_CAPACITY
+/// Zero fails validation, so the value comes from the embedded `config.toml`.
+fn default_client_queue_capacity() -> usize {
+    SERVER_CONFIG.message_bus.client_queue_capacity as usize
+}
+
+/// [`IggyByteSize`]'s own `Default` is 0 bytes, which turns the read-ahead
+/// buffer off, so the value comes from the embedded `config.toml`.
+fn default_replica_read_buffer_size() -> IggyByteSize {
+    SERVER_CONFIG
+        .message_bus
+        .replica_read_buffer_size
+        .parse()
+        .expect("message_bus.replica_read_buffer_size is a byte size")
 }
 
 #[cfg(test)]
@@ -241,7 +278,10 @@ mod tests {
             .remove("client_queue_capacity");
         config["peer_queue_capacity"] = serde_json::json!(8192);
         let decoded: MessageBusConfig = serde_json::from_value(config).unwrap();
-        assert_eq!(decoded.client_queue_capacity, DEFAULT_CLIENT_QUEUE_CAPACITY);
+        assert_eq!(
+            decoded.client_queue_capacity,
+            baseline().client_queue_capacity
+        );
         assert_eq!(decoded.peer_queue_capacity, 8192);
         decoded.validate().unwrap();
     }
@@ -253,6 +293,62 @@ mod tests {
         assert!(config.validate().is_err());
     }
 
+    #[test]
+    fn default_replica_read_buffer_is_256_kib() {
+        assert_eq!(
+            baseline().replica_read_buffer_size.as_bytes_u64(),
+            256 * 1024
+        );
+    }
+
+    /// Zero is the unbuffered path, not a misconfiguration: it is the
+    /// A/B baseline arm and must survive validation.
+    #[test]
+    fn accepts_zero_replica_read_buffer() {
+        let mut c = baseline();
+        c.replica_read_buffer_size = IggyByteSize::from(0_u64);
+        assert!(c.validate().is_ok());
+    }
+
+    #[test]
+    fn accepts_replica_read_buffer_at_floor() {
+        let mut c = baseline();
+        c.replica_read_buffer_size = IggyByteSize::from(MIN_REPLICA_READ_BUFFER_BYTES);
+        assert!(c.validate().is_ok());
+    }
+
+    #[test]
+    fn rejects_replica_read_buffer_below_floor() {
+        let mut c = baseline();
+        c.replica_read_buffer_size = IggyByteSize::from(MIN_REPLICA_READ_BUFFER_BYTES - 1);
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn replica_read_buffer_parses_byte_size_strings() {
+        for (text, expected) in [("0", 0_u64), ("64 KiB", 64 * 1024)] {
+            let mut config = serde_json::to_value(baseline()).unwrap();
+            config["replica_read_buffer_size"] = serde_json::json!(text);
+            let decoded: MessageBusConfig = serde_json::from_value(config).unwrap();
+            assert_eq!(decoded.replica_read_buffer_size.as_bytes_u64(), expected);
+            decoded.validate().unwrap();
+        }
+    }
+
+    #[test]
+    fn missing_replica_read_buffer_keeps_config_default() {
+        let mut config = serde_json::to_value(baseline()).unwrap();
+        config
+            .as_object_mut()
+            .unwrap()
+            .remove("replica_read_buffer_size");
+        let decoded: MessageBusConfig = serde_json::from_value(config).unwrap();
+        assert_eq!(
+            decoded.replica_read_buffer_size,
+            baseline().replica_read_buffer_size
+        );
+    }
+
     #[test]
     fn rejects_zero_max_message_size() {
         let mut c = baseline();

core/integration/tests/cluster/mod.rs

@@ -31,4 +31,5 @@ mod partition_dedup;
 mod partition_primary_routing;
 mod partition_state_transfer;
 mod register_forwarding;
+mod replica_read_batching;
 mod staggered_bootstrap;

core/integration/tests/cluster/replica_read_batching.rs

@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The plaintext replica plane counts what it reads.
+//!
+//! `replica_socket_reads_total` and `replica_inbound_frames_total` are the
+//! direct evidence for `message_bus.replica_read_buffer_size`: their ratio is
+//! the batching factor the read-ahead buffer exists to raise. Only a shard
+//! that owns a replica socket bumps them, so a nonzero pair also names the
+//! link shard.
+//!
+//! One shard per node, so shard 0 is that link shard and the assertion needs
+//! no search. What is pinned here is that both counters reach the scrape at
+//! all: the `Rc` reaches the reader task, the take in `tick_partitions` runs,
+//! and the two `ShardMetrics` counters are registered. The ratio itself is
+//! only logged. Frame arrival timing on a live link is not controlled, so a
+//! threshold on it would be a flaky test;
+//! `framing::tests::buffered_read_batches_many_frames_per_socket_read` asserts
+//! the factor where the traffic shape is fixed.
+//!
+//! `frames >= reads` is deliberately not asserted: a body split across two
+//! segments, or one straddling the end of the buffer, costs two reads for one
+//! frame.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+use crate::server::http_client::HttpClient;
+
+const STREAM: &str = "replica-read-stream";
+const TOPIC: &str = "replica-read-topic";
+const PARTITION_ID: u32 = 0;
+const MESSAGES: u32 = 64;
+
+/// The counters land on the scrape through the shard's partition tick, so the
+/// first scrape after a produce can still read zero.
+const COUNTER_BUDGET: Duration = Duration::from_secs(10);
+const COUNTER_POLL: Duration = Duration::from_millis(250);
+
+/// One shard per node, so this is the only shard and it owns both replica
+/// links.
+const LINK_SHARD: u16 = 0;
+
+/// Read one `shard`-labelled counter out of the Prometheus text exposition.
+///
+/// The sub-registry label comes first in the label set, and these two counters
+/// carry no others, so the series name plus the label is an exact line prefix.
+fn shard_counter(metrics: &str, name: &str, shard: u16) -> Option<u64> {
+    let prefix = format!("{name}{{shard=\"{shard}\"}} ");
+    metrics
+        .lines()
+        .find_map(|line| line.strip_prefix(&prefix)?.trim().parse().ok())
+}
+
+async fn scrape(http: &HttpClient) -> String {
+    http.client
+        .get(http.url("/metrics"))
+        .bearer_auth(&http.token)
+        .send()
+        .await
+        .expect("metrics response")
+        .text()
+        .await
+        .expect("metrics text")
+}
+
+#[iggy_harness(cluster_nodes = 3, server(sharding.cpu_allocation = "0..1"))]
+async fn given_a_replicating_cluster_when_scraping_should_report_replica_reads_and_frames(
+    harness: &TestHarness,
+) {
+    let client = harness.new_client().await.unwrap();
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .unwrap();
+    let stream = Identifier::named(STREAM).unwrap();
+    let topic = Identifier::named(TOPIC).unwrap();
+    client.create_stream(STREAM).await.unwrap();
+    client
+        .create_topic(
+            &stream,
+            TOPIC,
+            &TopicCreateOptions {
+                partitions_count: Some(1),
+                message_expiry: Some(IggyExpiry::NeverExpire),
+                ..TopicCreateOptions::default()
+            },
+        )
+        .await
+        .expect("create topic");
+
+    for index in 0..MESSAGES {
+        let mut messages = vec![
+            IggyMessage::builder()
+                .payload(format!("payload-{index}").into())
+                .build()
+                .expect("message build"),
+        ];
+        client
+            .send_messages(
+                &stream,
+                &topic,
+                &Partitioning::partition_id(PARTITION_ID),
+                &mut messages,
+            )
+            .await
+            .unwrap_or_else(|error| panic!("send_messages {index}: {error}"));
+    }
+
+    let http = HttpClient::login_root(harness).await;
+    let deadline = Instant::now() + COUNTER_BUDGET;
+    loop {
+        let metrics = scrape(&http).await;
+        let reads = shard_counter(&metrics, "replica_socket_reads_total", LINK_SHARD);
+        let frames = shard_counter(&metrics, "replica_inbound_frames_total", LINK_SHARD);
+        if let (Some(reads), Some(frames)) = (reads, frames)
+            && reads > 0
+            && frames > 0
+        {
+            println!(
+                "link shard {LINK_SHARD}: {frames} frames over {reads} socket reads, \
+                 {:.2} frames per read",
+                frames as f64 / reads as f64
+            );
+            return;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "link shard {LINK_SHARD} reported reads={reads:?} frames={frames:?} within \
+             {COUNTER_BUDGET:?}; both must be nonzero once replica traffic has flowed"
+        );
+        sleep(COUNTER_POLL).await;
+    }
+}

core/message_bus/src/config.rs

@@ -138,6 +138,12 @@ pub struct MessageBusConfig {
     /// validator; undersize or oversize frames are rejected.
     pub max_message_size: usize,
 
+    /// Read-ahead buffer per plaintext replica link, in bytes. Threaded
+    /// into `TcpTransportConn::with_replica_read` by the replica
+    /// installer; zero selects the unbuffered read path. The client
+    /// plane and the TLS-family transports ignore it.
+    pub replica_read_buffer_size: usize,
+
     /// Bound on each replica peer's mpsc queue. The writer task drains; the
     /// `send_to_*` path enqueues. Too small drops under burst; too
     /// large delays backpressure signalling.
@@ -216,6 +222,8 @@ impl From<&ServerConfig> for MessageBusConfig {
             max_batch: bus.max_batch,
             max_message_size: usize::try_from(bus.max_message_size.as_bytes_u64())
                 .expect("message_bus.max_message_size fits usize on supported targets"),
+            replica_read_buffer_size: usize::try_from(bus.replica_read_buffer_size.as_bytes_u64())
+                .expect("message_bus.replica_read_buffer_size fits usize on supported targets"),
             peer_queue_capacity: bus.peer_queue_capacity,
             client_queue_capacity: bus.client_queue_capacity,
             reconnect_period: bus.reconnect_period.get_duration(),