v1.1.0 — Gzip canvas compression + throttled sends

- Canvas snapshots gzip-compressed before sending: 1MB raw -> ~1-5KB
  for a mostly-white canvas, ~100-500KB for a busy one
- Semaphore throttles concurrent canvas sends (default 20) to prevent
  memory spikes during connection bursts
- Frontend decompresses gzip via browser DecompressionStream API
- New config: MAX_CONCURRENT_CANVAS_SENDS (default 20)
- Added flate2 dependency for gzip encoding

crates/adapters/config-env/src/lib.rs

@@ -14,6 +14,7 @@ const DEFAULT_BROADCAST_CAPACITY: usize = 1024;
 const DEFAULT_SNAPSHOT_INTERVAL_SECS: u64 = 300;
 const DEFAULT_SNAPSHOT_MAX: usize = 5;
 const DEFAULT_SNAPSHOT_DIR: &str = "snapshots/";
+const DEFAULT_MAX_CONCURRENT_CANVAS_SENDS: usize = 20;
 
 pub struct EnvConfigSource;
 
@@ -24,6 +25,10 @@ impl ConfigSource for EnvConfigSource {
                 address: env_or("ADDRESS", DEFAULT_ADDRESS),
                 port: parse_env("PORT", DEFAULT_PORT)?,
                 enable_cors: parse_bool_env("ENABLE_CORS", true),
+                max_concurrent_canvas_sends: parse_env(
+                    "MAX_CONCURRENT_CANVAS_SENDS",
+                    DEFAULT_MAX_CONCURRENT_CANVAS_SENDS,
+                )?,
             },
             canvas: CanvasConfig {
                 width: parse_env("CANVAS_WIDTH", DEFAULT_CANVAS_WIDTH)?,

crates/adapters/websocket/Cargo.toml

@@ -7,6 +7,7 @@ edition.workspace = true
 application = { workspace = true }
 domain = { workspace = true }
 axum = { workspace = true, features = ["ws"] }
+flate2 = { workspace = true }
 futures = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

crates/adapters/websocket/src/handler.rs

@@ -1,3 +1,4 @@
+use std::io::Write;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
 
@@ -5,8 +6,10 @@ use axum::extract::State;
 use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
 use axum::response::IntoResponse;
 use domain::{BroadcastEvent, BroadcastSubscription, Color, Position};
+use flate2::Compression;
+use flate2::write::GzEncoder;
 use futures::{SinkExt, StreamExt, stream::SplitSink};
-use tokio::sync::mpsc;
+use tokio::sync::{Semaphore, mpsc};
 use tracing::{error, info};
 
 use crate::WsState;
@@ -36,7 +39,7 @@ async fn handle_connection(socket: WebSocket, state: Arc<WsState>) {
     // Subscribe before snapshotting to avoid missing updates
     let subscription = state.app_state.broadcaster().subscribe();
 
-    if !send_canvas_snapshot(&mut sender, &state.app_state).await {
+    if !send_canvas_snapshot(&mut sender, &state.app_state, &state.canvas_send_semaphore).await {
         return;
     }
 
@@ -67,10 +70,39 @@ async fn handle_connection(socket: WebSocket, state: Arc<WsState>) {
     application::soldiers::disconnect::execute(&state.app_state, &connection_id);
 }
 
-async fn send_canvas_snapshot(sender: &mut WsSender, state: &AppState) -> bool {
+async fn send_canvas_snapshot(
+    sender: &mut WsSender,
+    state: &AppState,
+    semaphore: &Semaphore,
+) -> bool {
+    let Ok(_permit) = semaphore.acquire().await else {
+        return false;
+    };
+
     let pixels = application::canvas::get_state::execute(state);
-    let bytes = Color::collect_as_bytes(&pixels);
-    sender.send(Message::Binary(bytes.into())).await.is_ok()
+    let raw_bytes = Color::collect_as_bytes(&pixels);
+
+    let Some(compressed) = gzip_compress(&raw_bytes) else {
+        error!("Failed to compress canvas snapshot");
+        return false;
+    };
+
+    info!(
+        "Sending canvas snapshot: {}KB raw -> {}KB gzip",
+        raw_bytes.len() / 1024,
+        compressed.len() / 1024
+    );
+
+    sender
+        .send(Message::Binary(compressed.into()))
+        .await
+        .is_ok()
+}
+
+fn gzip_compress(data: &[u8]) -> Option<Vec<u8>> {
+    let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
+    encoder.write_all(data).ok()?;
+    encoder.finish().ok()
 }
 
 async fn run_send_loop(

crates/adapters/websocket/src/lib.rs

@@ -6,16 +6,19 @@ use std::sync::atomic::AtomicU64;
 
 use application::AppState;
 use axum::{Router, routing::get};
+use tokio::sync::Semaphore;
 
 pub(crate) struct WsState {
     app_state: Arc<AppState>,
     connection_counter: AtomicU64,
+    canvas_send_semaphore: Arc<Semaphore>,
 }
 
-pub fn build_router(state: Arc<AppState>) -> Router {
+pub fn build_router(state: Arc<AppState>, max_concurrent_canvas_sends: usize) -> Router {
     let ws_state = Arc::new(WsState {
         app_state: state,
         connection_counter: AtomicU64::new(0),
+        canvas_send_semaphore: Arc::new(Semaphore::new(max_concurrent_canvas_sends)),
     });
 
     Router::new()

crates/config/src/lib.rs

@@ -24,6 +24,7 @@ pub struct ServerConfig {
     pub address: String,
     pub port: u16,
     pub enable_cors: bool,
+    pub max_concurrent_canvas_sends: usize,
 }
 
 #[derive(Debug, Clone)]

crates/server/src/main.rs

@@ -127,7 +127,7 @@ fn build_app(
     state: Arc<AppState>,
     config: &AppConfig,
 ) -> Result<axum::Router, Box<dyn std::error::Error>> {
-    let ws_router = websocket::build_router(state);
+    let ws_router = websocket::build_router(state, config.server.max_concurrent_canvas_sends);
     let http_router = http_axum::build_router(config.server.enable_cors, &config.rate_limit)?;
     Ok(ws_router.merge(http_router))
 }