update deps, improve payload handling using Bytes O(N) -> O(1), and expose some configs to public

Author: ronakgh97

.github/workflows/ci.yml

@@ -48,4 +48,4 @@ jobs:
         uses: Swatinem/rust-cache@v2
 
       - name: Run tests
-        run: cargo nextest run --all-targets --no-fail-fast
+        run: cargo nextest run --all-targets

Cargo.toml

@@ -10,12 +10,15 @@ homepage = "https:/github.com/ronakgh97/ghost-sync"
 documentation = "https:/github.com/ronakgh97/ghost-sync/master/README.md"
 
 [dependencies]
-tokio = { version = "1.50.0", features = ["full", "test-util"] }
+tokio = { version = "1.50.0", features = ["full"] }
 dashmap = "6.1.0"
 anyhow = "1.0.102"
 wincode = { version = "0.4.8", features = ["derive", "uuid"] }
 uuid = { version = "1.22.0", features = ["v4"] }
 bytes = "1.11.1"
 colored = "3.1.1"
 dotenv = "0.15.0"
-chrono = "0.4.44"
+chrono = "0.4.44"
+
+[dev-dependencies]
+tokio = { version = "1.50.0", features = ["test-util"] }

src/client.rs

@@ -49,7 +49,7 @@ impl Client {
     }
 
     /// Receive the next server event.
-    /// Ping/pong keepalive is handled internally — the user never sees it.
+    /// Ping/pong keepalive is handled internally
     /// Returns `Ok(None)` on clean disconnect.
     pub async fn recv(&mut self) -> Result<Option<ServerEvent>, SyncError> {
         loop {
@@ -104,7 +104,7 @@ pub struct ClientBuilder {
 impl ClientBuilder {
     pub fn new() -> Self {
         Self {
-            max_payload: 64 * 1024,
+            max_payload: 256 * 1024,
         }
     }
 }
@@ -134,8 +134,8 @@ impl ClientBuilder {
         let (read_half, write_half) = stream.into_split();
 
         Ok(Client {
-            reader: BufReader::new(read_half),
-            writer: BufWriter::new(write_half),
+            reader: BufReader::with_capacity(64 * 1024, read_half),
+            writer: BufWriter::with_capacity(64 * 1024, write_half),
             max_payload: self.max_payload,
         })
     }

src/room.rs

@@ -1,10 +1,13 @@
-use crate::types::{Result, ServerWire, SyncError};
+use bytes::Bytes;
 use dashmap::DashMap;
 use tokio::sync::mpsc;
 use uuid::Uuid;
 
-/// A broadcast frame to be sent to a client.
-pub type BroadcastFrame = Vec<u8>;
+use crate::types::{Result, ServerWire, SyncError};
+
+/// A broadcast frame to be sent to a client. `Bytes` is reference-counted,
+/// so cloning for multiple recipients is an Arc bump, not a memcpy.
+pub type BroadcastFrame = Bytes;
 
 /// A single room holding connected clients.
 ///
@@ -46,19 +49,21 @@ impl Room {
     #[inline(always)]
     pub async fn broadcast(&self, sender: Uuid, msg: &ServerWire) -> Vec<Uuid> {
         let payload = match wincode::serialize(msg) {
-            Ok(p) => p,
+            Ok(p) => Bytes::from(p),
             Err(_) => return Vec::new(),
         };
-        self.broadcast_raw(sender, &payload).await
+        self.broadcast_raw(sender, payload).await
     }
 
-    /// Send raw bytes to all clients except `sender`.
+    /// Send pre-serialized bytes to all clients except `sender`.
     /// Returns the UUIDs of clients whose write channels were full (frame dropped).
+    ///
+    /// Cloning `Bytes` is an Arc pointer bump — no data copy regardless of payload size.
     #[inline(always)]
-    pub async fn broadcast_raw(&self, sender: Uuid, payload: &[u8]) -> Vec<Uuid> {
+    pub async fn broadcast_raw(&self, sender: Uuid, payload: Bytes) -> Vec<Uuid> {
         let mut dropped = Vec::new();
         for entry in self.clients.iter() {
-            if *entry.key() != sender && entry.value().try_send(payload.to_vec()).is_err() {
+            if *entry.key() != sender && entry.value().try_send(payload.clone()).is_err() {
                 dropped.push(*entry.key());
             }
         }
@@ -69,11 +74,9 @@ impl Room {
     #[inline(always)]
     pub async fn send_to(&self, id: &Uuid, msg: &ServerWire) {
         if let Some(tx) = self.clients.get(id) {
-            let payload = match wincode::serialize(msg) {
-                Ok(p) => p,
-                Err(_) => return,
-            };
-            let _ = tx.try_send(payload);
+            if let Ok(payload) = wincode::serialize(msg) {
+                let _ = tx.try_send(Bytes::from(payload));
+            }
         }
     }
 
@@ -90,7 +93,7 @@ pub struct RoomManager {
 }
 
 impl RoomManager {
-    /// Init a room with 64s pre-allocated
+    /// Init a room manager with 64 pre-allocated slots.
     pub fn new() -> Self {
         Self {
             rooms: DashMap::with_capacity(64),

src/server.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use bytes::Bytes;
 use dashmap::DashMap;
 use tokio::io::BufWriter;
 use tokio::net::{TcpListener, TcpStream};
@@ -52,7 +53,7 @@ pub struct Server {
 /// Handle to a running server.
 ///
 /// Provides runtime control: shutdown and room management.
-/// Rooms can be created and deleted at any time — new clients will see the
+/// Rooms can be created and deleted at any time, new clients will see the
 /// updated room list, and existing clients in deleted rooms get errors on
 /// their next broadcast or join attempt.
 pub struct ServerHandle {
@@ -102,18 +103,13 @@ impl Server {
     }
 
     /// Create a room. Call this before [`Server::run`].
-    ///
     /// Clients can only join rooms that have been explicitly created.
-    ///
-    /// # Errors
-    ///
     /// Returns [`SyncError::RoomAlreadyExists`] if a room with this ID exists.
     pub fn create_room(&self, id: &str) -> Result<()> {
         self.rooms.create(id)
     }
 
     /// Delete a room. Call this before [`Server::run`].
-    ///
     /// Returns `true` if the room existed.
     pub fn delete_room(&self, id: &str) -> bool {
         self.rooms.delete(id)
@@ -122,7 +118,7 @@ impl Server {
     /// Start accepting connections.
     ///
     /// Binds the TCP listener synchronously. If the bind fails, the error is
-    /// returned immediately — nothing is spawned on failure.
+    /// returned immediately
     ///
     /// On success, spawns the accept loop on the current runtime and returns
     /// a [`ServerHandle`] for shutdown and runtime action
@@ -188,14 +184,14 @@ impl Server {
         Ok(handle)
     }
 
-    #[inline(always)]
+    #[inline]
     async fn handle_connection(self: &Arc<Self>, stream: TcpStream) -> Result<()> {
         let client_id = Uuid::new_v4();
         let (read_half, write_half) = stream.into_split();
         let mut reader = tokio::io::BufReader::new(read_half);
         let mut writer = BufWriter::new(write_half);
 
-        let (write_tx, mut write_rx) = mpsc::channel::<Vec<u8>>(64);
+        let (write_tx, mut write_rx) = mpsc::channel::<Bytes>(self.config.channel_capacity);
 
         // Spawn writer task
         let writer_handle = tokio::spawn(async move {
@@ -222,11 +218,12 @@ impl Server {
         result
     }
 
+    #[inline(always)]
     async fn client_loop(
         self: &Arc<Self>,
         client_id: Uuid,
         reader: &mut tokio::io::BufReader<tokio::net::tcp::OwnedReadHalf>,
-        write_tx: &mpsc::Sender<Vec<u8>>,
+        write_tx: &mpsc::Sender<Bytes>,
     ) -> Result<()> {
         let mut shutdown_rx = self.shutdown_tx.subscribe();
         let mut ping_interval = tokio::time::interval(self.config.ping_interval);
@@ -289,7 +286,7 @@ impl Server {
         self: &Arc<Self>,
         client_id: Uuid,
         msg: ClientWire,
-        write_tx: &mpsc::Sender<Vec<u8>>,
+        write_tx: &mpsc::Sender<Bytes>,
     ) -> Result<()> {
         match msg {
             ClientWire::JoinRoom { room_id } => {
@@ -375,9 +372,10 @@ impl Server {
                 };
                 let payload = wincode::serialize(&broadcast)
                     .map_err(|e| SyncError::Protocol(format!("serialize failed: {:?}", e)))?;
+                let payload = Bytes::from(payload);
 
                 if let Some(room) = self.rooms.get(&room_id) {
-                    let dropped = room.broadcast_raw(client_id, &payload).await;
+                    let dropped = room.broadcast_raw(client_id, payload).await;
                     for id in dropped {
                         self.handler.on_backpressure(id, &room_id);
                     }
@@ -392,6 +390,7 @@ impl Server {
         Ok(())
     }
 
+    #[inline]
     async fn cleanup_client(self: &Arc<Self>, client_id: Uuid, room_id: &str) {
         let notify = ServerWire::PlayerLeft { client_id };
         if let Some(room) = self.rooms.get(room_id) {
@@ -407,9 +406,9 @@ impl Server {
     }
 
     #[inline(always)]
-    async fn send_to_client(&self, client_id: Uuid, tx: &mpsc::Sender<Vec<u8>>, msg: &ServerWire) {
+    async fn send_to_client(&self, client_id: Uuid, tx: &mpsc::Sender<Bytes>, msg: &ServerWire) {
         if let Ok(payload) = wincode::serialize(msg) {
-            if tx.try_send(payload).is_err() {
+            if tx.try_send(Bytes::from(payload)).is_err() {
                 if let Some(state) = self.clients.get(&client_id) {
                     if let Some(ref room_id) = state.room_id {
                         self.handler.on_backpressure(client_id, room_id);
@@ -484,6 +483,14 @@ impl ServerBuilder {
         self
     }
 
+    /// Per-client write channel capacity. Frames are dropped when full
+    /// (with `on_backpressure` hook). Higher values buffer more for bursty
+    /// games; lower values keep latency tight. Default: 64.
+    pub fn channel_capacity(mut self, n: usize) -> Self {
+        self.config.channel_capacity = n;
+        self
+    }
+
     /// Set a custom event handler for lifecycle hooks.
     pub fn handler(mut self, h: impl ServerHandler) -> Self {
         self.handler = Arc::new(h);

src/types.rs

@@ -50,6 +50,10 @@ pub struct ServerConfig {
     /// How often the server pings clients. If no Pong arrives before the
     /// next ping tick, the client is disconnected with [`SyncError::PingTimeout`].
     pub ping_interval: Duration,
+    /// Per-client write channel capacity. Frames are dropped when the channel
+    /// is full (with `on_backpressure` hook). Higher values buffer more for
+    /// bursty games; lower values keep latency tight.
+    pub channel_capacity: usize,
 }
 
 impl Default for ServerConfig {
@@ -61,6 +65,7 @@ impl Default for ServerConfig {
             // Non-divisible defaults to avoid idle_timeout and ping_interval
             idle_timeout: Duration::from_secs(31),
             ping_interval: Duration::from_secs(13),
+            channel_capacity: 64,
         }
     }
 }