Main

src/index.js

@@ -32,8 +32,11 @@
 }
 
 let wss;
+let httpServer;
+let markShuttingDown;
+let sessionManager;
 try {
-  ({ wss } = createServer(config));
+  ({ wss, httpServer, markShuttingDown, sessionManager } = createServer(config));
 } catch (err) {
   logger.error("Failed to start server", { error: err.message });
   process.exit(1);
@@ -128,8 +131,16 @@
   });
 }
 
-process.on("SIGTERM", () => shutdown(wss, "SIGTERM"));
-process.on("SIGINT", () => shutdown(wss, "SIGINT"));
+process.on("SIGTERM", () => {
+  markShuttingDown();
+  wss.close();
+  shutdown(httpServer, "SIGTERM");
+});
+process.on("SIGINT", () => {
+  markShuttingDown();
+  wss.close();
+  shutdown(httpServer, "SIGINT");
+});
 
 process.on("uncaughtException", (err) => {
   logger.error("Uncaught exception", { error: err.message });

src/room-manager.js

@@ -1,12 +1,24 @@
 import { WebSocket } from "ws";
 
+/**
+ * @typedef {Object} BackpressureOptions
+ * @property {boolean} [enabled=false] - Enable backpressure-aware broadcasting
+ * @property {number} [highWaterMark=1048576] - Bytes threshold to flag client as slow (default 1MB)
+ * @property {number} [slowConsumerTimeout=30000] - Ms before terminating slow consumer (default 30s)
+ * @property {number} [batchSize=100] - Number of sends per event loop tick (default 100)
+ */
+
 /**
  * Manages room membership and message broadcasting for connected WebSocket clients.
  *
  * Rooms are keyed by an arbitrary string ID. Each room holds a map of
  * `clientId → WebSocket` so broadcasts are O(members). A reverse index
  * (`_clientRooms`) enables O(1) lookup of all rooms a client belongs to,
  * which is used during disconnection cleanup.
+ *
+ * When backpressure options are provided, broadcast() operates in a
+ * non-blocking batched mode with per-client slow consumer detection,
+ * message coalescing, and automatic eviction.
  */
 export class RoomManager {
   constructor({ maxRoomSize = Infinity } = {}) {
@@ -33,6 +45,18 @@ export class RoomManager {
     return this._clientRooms.get(clientId);
   }
 
+  /** @private */
+  _ensureClientState(clientId, ws) {
+    if (!this._clientState.has(clientId)) {
+      this._clientState.set(clientId, {
+        ws,
+        slowSince: null,
+        coalescedMessage: null,
+      });
+    }
+    return this._clientState.get(clientId);
+  }
+
   /** @private */
   _cleanupRoom(roomId) {
     const room = this._rooms.get(roomId);
@@ -123,6 +147,13 @@ export class RoomManager {
       clientRooms.delete(roomId);
       this._cleanupClient(clientId);
     }
+
+    if (this._backpressureOptions.enabled) {
+      const clientRoomsRemaining = this._clientRooms.get(clientId);
+      if (!clientRoomsRemaining || clientRoomsRemaining.size === 0) {
+        this._cleanupClientState(clientId);
+      }
+    }
   }
 
   /**
@@ -181,17 +212,118 @@ export class RoomManager {
     const room = this._rooms.get(roomId);
     if (!room) return;
 
-    const data = typeof message === "string" ? message : JSON.stringify(message);
+    if (!this._backpressureOptions.enabled) {
+      const data = typeof message === "string" ? message : JSON.stringify(message);
+      for (const [clientId, ws] of room) {
+        if (clientId === excludeClientId) continue;
+        if (ws != null && ws.readyState === WebSocket.OPEN) {
+          try {
+            ws.send(data);
+          } catch {
+            // ignore send errors for individual clients
+          }
+        }
+      }
+      return;
+    }
+
+    this._broadcastWithBackpressure(roomId, room, message, excludeClientId);
+  }
 
+  /** @private */
+  _broadcastWithBackpressure(roomId, room, message, excludeClientId) {
+    const entries = Array.from(room.entries());
+    const batchSize = this._backpressureOptions.batchSize;
+    let index = 0;
+
+    const processBatch = () => {
+      const end = Math.min(index + batchSize, entries.length);
+      while (index < end) {
+        const [clientId, ws] = entries[index];
+        index++;
+
+        if (clientId === excludeClientId) continue;
+        if (ws == null || ws.readyState !== WebSocket.OPEN) continue;
+
+        const state = this._clientState.get(clientId);
+        const bufferedAmount = ws.bufferedAmount || 0;
+        const isSlow = state && state.slowSince !== null;
+        const exceedsHighWater = bufferedAmount > this._backpressureOptions.highWaterMark;
+
+        if (exceedsHighWater && state && state.slowSince === null) {
+          state.slowSince = Date.now();
+          this._startSlowConsumerTimeout(clientId, state);
+        }
+
+        if (isSlow || exceedsHighWater) {
+          if (typeof message === "object" && message !== null && message.type === "location_update") {
+            if (state) {
+              state.coalescedMessage = message;
+            }
+          } else {
+            this._sendToClient(clientId, ws, message);
+          }
+        } else {
+          this._sendToClient(clientId, ws, message);
+        }
+      }
+
+      if (index < entries.length) {
+        setImmediate(processBatch);
+      } else {
+        this._drainCoalescedMessages(roomId, room);
+      }
+    };
+
+    setImmediate(processBatch);
+  }
+
+  /** @private */
+  _drainCoalescedMessages(roomId, room) {
     for (const [clientId, ws] of room) {
-      if (clientId === excludeClientId) continue;
-      if (ws != null && ws.readyState === WebSocket.OPEN) {
-        try {
-          ws.send(data);
-        } catch {
-          // ignore send errors for individual clients
+      if (ws == null || ws.readyState !== WebSocket.OPEN) continue;
+
+      const state = this._clientState.get(clientId);
+      if (state && state.coalescedMessage !== null) {
+        const msg = state.coalescedMessage;
+        state.coalescedMessage = null;
+        this._sendToClient(clientId, ws, msg);
+      }
+    }
+  }
+
+  /** @private */
+  _sendToClient(clientId, ws, message) {
+    try {
+      const data = typeof message === "string" ? message : JSON.stringify(message);
+      ws.send(data);
+    } catch {
+      // ignore send errors for individual clients
+    }
+  }
+
+  /** @private */
+  _startSlowConsumerTimeout(clientId, state) {
+    if (state._timeoutId) {
+      clearTimeout(state._timeoutId);
+    }
+
+    state._timeoutId = setTimeout(() => {
+      if (state.slowSince !== null) {
+        const ws = state.ws;
+        if (ws && ws.readyState === WebSocket.OPEN) {
+          try {
+            ws.close(4000, "Slow consumer");
+          } catch {
+            // ignore close errors
+          }
         }
+        this._cleanupClientState(clientId);
       }
+    }, this._backpressureOptions.slowConsumerTimeout);
+
+    if (state._timeoutId.unref) {
+      state._timeoutId.unref();
     }
   }
 
@@ -283,6 +415,10 @@ export class RoomManager {
       }
       this._clientRooms.delete(clientId);
     }
+
+    if (this._backpressureOptions.enabled) {
+      this._cleanupClientState(clientId);
+    }
   }
 
   /**
@@ -311,6 +447,47 @@ export class RoomManager {
     return rooms ? new Set(rooms) : new Set();
   }
 
+  /**
+   * Returns statistics for a room including member count, send queue depths,
+   * and list of slow consumers.
+   *
+   * @param {string} roomId - Identifier of the room to query.
+   * @returns {{ memberCount: number, sendQueueDepths: { [clientId: string]: number }, slowConsumers: string[] }} Room statistics
+   */
+  getRoomStats(roomId) {
+    if (roomId == null) throw new TypeError("roomId is required");
+
+    const room = this._rooms.get(roomId);
+    if (!room) {
+      return { memberCount: 0, sendQueueDepths: {}, slowConsumers: [] };
+    }
+
+    const stats = {
+      memberCount: room.size,
+      sendQueueDepths: {},
+      slowConsumers: [],
+    };
+
+    if (!this._backpressureOptions.enabled) {
+      return stats;
+    }
+
+    for (const [clientId, ws] of room) {
+      if (ws != null && ws.readyState === WebSocket.OPEN) {
+        stats.sendQueueDepths[clientId] = ws.bufferedAmount || 0;
+      } else {
+        stats.sendQueueDepths[clientId] = 0;
+      }
+
+      const state = this._clientState.get(clientId);
+      if (state && state.slowSince !== null) {
+        stats.slowConsumers.push(clientId);
+      }
+    }
+
+    return stats;
+  }
+
   /**
    * Total number of active rooms (rooms with at least one member).
    * @type {number}