chore: enable nodelay on replicas

crates/fluvio-service/src/server.rs

@@ -128,6 +128,10 @@ where
             .unwrap_or_else(|_| "".to_owned());
         debug!(%peer_addr, "Handling request");
 
+        stream.set_nodelay(true).unwrap_or_else(|err| {
+            error!("Error setting nodelay: {}", err);
+        });
+
         let socket = {
             let fd = stream.as_raw_fd();
             FluvioSocket::from_stream(Box::new(stream.clone()), Box::new(stream), fd)

crates/fluvio-socket/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-socket"
-version = "0.15.2"
+version = "0.15.3"
 edition = "2021"
 authors = ["Fluvio Contributors <[email protected]>"]
 description = "Provide TCP socket wrapper for fluvio protocol"
@@ -31,6 +31,7 @@ pin-project = { workspace = true }
 thiserror = { workspace = true }
 semver = { workspace = true }
 nix = { workspace = true, features = ["uio"]}
+libc = { workspace = true }
 
 # Fluvio dependencies
 fluvio-future = { workspace = true, features = ["net", "task", "retry"] }

crates/fluvio-socket/src/socket.rs

@@ -119,6 +119,23 @@ impl FluvioSocket {
         let connector = DefaultDomainConnector::new();
         Self::connect_with_connector(addr, &connector).await
     }
+
+    #[cfg(not(target_arch = "wasm32"))]
+    pub fn set_nodelay(&self, value: bool) {
+        let fd = self.id();
+        let opt = libc::IPPROTO_TCP;
+        let val = libc::TCP_NODELAY;
+        let value: i32 = if value { 1 } else { 0 };
+        let payload = std::ptr::addr_of!(value).cast();
+        unsafe {
+            libc::setsockopt(fd, opt, val, payload, std::mem::size_of::<u32>() as u32);
+        }
+    }
+
+    #[cfg(target_arch = "wasm32")]
+    pub fn set_nodelay(&self, _value: bool) {
+        // No-op for WASM
+    }
 }
 
 cfg_if::cfg_if! {

crates/fluvio-spu/src/replication/follower/controller.rs

@@ -320,6 +320,7 @@ mod inner {
 
                 match FluvioSocket::connect(&leader_endpoint).await {
                     Ok(mut socket) => {
+                        socket.set_nodelay(true);
                         debug!("connected to leader");
 
                         match self.send_fetch_stream_request(&mut socket).await {