diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f3171e..c04f2d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +- Reclaimed empty rooms on last leave and on disconnect, announced them with the new `ROOM_REMOVED` + frame, and taught the console and Swing clients to drop reclaimed rooms. +- Bounded distinct rooms with a configurable `maxRooms` limit (default 256, CLI `--max-rooms`) and + answered over-limit `ROOM_JOIN` with an explicit `ERROR`. +- Added per-connection token buckets for inbound frames and room creation that reply `ERROR` instead + of dropping the connection. + ## 1.6.1 - 2026-06-19 - Hardened account storage, history migration, TLS socket handling, and integration coverage after repository audit. diff --git a/docs/architecture.md b/docs/architecture.md index ca5b43b..27d3810 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -22,6 +22,8 @@ clients <-> ChatConnection <-> ChatProtocol <-> ChatServer room history is replayed after `ROOM_JOINED`. 9. Admin users can send `/health` to receive a private server status frame. 10. Closing or failed connections are removed and announced with `USER_REMOVED`. +11. A non-default room is reclaimed as soon as its last member leaves or disconnects, and the + reclamation is announced with `ROOM_REMOVED`. ## Message Frames @@ -34,7 +36,7 @@ Frames are one-line UTF-8 JSON objects serialized by `ChatProtocol`. `PRIVATE_TEXT`. - `sender` carries the author for text frames; clients own display formatting. - `room` carries room routing for `ROOM_TEXT`, `ROOM_JOIN`, `ROOM_LEAVE`, `ROOM_ADDED`, - `ROOM_JOINED`, and `ROOM_LEFT`. + `ROOM_REMOVED`, `ROOM_JOINED`, and `ROOM_LEFT`. - `recipient` carries the target user for `PRIVATE_TEXT`. - Clients may supply `timestamp` and `messageId`, but the server replaces both when it normalizes a room or private message so clients cannot forge identity or ordering metadata. @@ -50,11 +52,15 @@ Frames are one-line UTF-8 JSON objects serialized by `ChatProtocol`. - `port` - TCP port, default `1500`. - `bindAddress` - listen address, default loopback-only `127.0.0.1`. - `maxClients` - maximum concurrent client handler threads, default `100`. +- `maxRooms` - maximum number of distinct rooms, default `256`, CLI `--max-rooms`. - `handshakeTimeout` - maximum time to complete username registration, default `10s`. - `readTimeout` - idle socket read timeout after handshake, default `5m`. - `historyFile` / `historyLimit` / `historyReplayLimit` - optional replayable JSONL history. - `accountFile` - optional account registry. - `tls` - optional JSSE TLS server socket configuration. +- `rateLimit` - per-connection token buckets for inbound frames (default burst `60`, `20/s`) and for + room creation (default burst `10`, `0.5/s`); exhausted buckets answer `ERROR` instead of closing + the connection. The legacy `new ChatServer(int port)` constructor delegates to `ChatServerConfig.ofPort(port)`. @@ -97,9 +103,13 @@ The Swing model stores a bounded local timeline for the current app session. Tex deduplicated by `messageId`, own messages are rendered as `Вы`, and user add/remove protocol events are appended as service events instead of ordinary chat text. -Room membership lives on the server. `general` always exists, room creation is idempotent through -`ROOM_JOIN`, distinct rooms are capped at 1,000, and clients that try to send to a room before -joining receive an explicit `ERROR`. +Room membership lives on the server. `general` always exists and is never reclaimed, room creation is +idempotent through `ROOM_JOIN`, distinct rooms are capped by `maxRooms`, and clients that try to send +to a room before joining receive an explicit `ERROR`. Every other room is reclaimed and announced +with `ROOM_REMOVED` once its last member leaves or disconnects, so room names cannot accumulate for +the lifetime of the process. Membership and reclamation run under a dedicated rooms monitor that +holds no I/O and is never nested inside the sessions monitor, so the cap stays exact and a room +cannot disappear between a joiner observing it and joining it. ## History Store diff --git a/src/integrationTest/java/dev/krotname/networkchat/NetworkChatIntegrationTest.java b/src/integrationTest/java/dev/krotname/networkchat/NetworkChatIntegrationTest.java index 71a5278..be5c680 100644 --- a/src/integrationTest/java/dev/krotname/networkchat/NetworkChatIntegrationTest.java +++ b/src/integrationTest/java/dev/krotname/networkchat/NetworkChatIntegrationTest.java @@ -1,6 +1,7 @@ package dev.krotname.networkchat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -141,6 +142,53 @@ void joiningAndLeavingRoomsProducesRoomEvents() throws Exception { alice.leaveRoom("dev"); awaitProtocolEvent(alice, MessageType.ROOM_LEFT, "dev"); + awaitProtocolEvent(alice, MessageType.ROOM_REMOVED, "dev"); + + assertFalse(server.getRooms().contains("dev")); + assertTrue(server.getRooms().contains(ChatMessage.GENERAL_ROOM)); + } + } + } + + @Test + void roomsAreReclaimedWhenTheirLastMemberDisconnects() throws Exception { + int port = 0; + try (ChatServer server = new ChatServer(port)) { + server.start(); + server.awaitStarted(); + port = server.getPort(); + + try (TestClient observer = new TestClient("observer", port)) { + observer.connect(); + try (TestClient alice = new TestClient("alice", port)) { + alice.connect(); + alice.joinRoom("dev"); + awaitProtocolEvent(observer, MessageType.ROOM_ADDED, "dev"); + } + + awaitProtocolEvent(observer, MessageType.ROOM_REMOVED, "dev"); + assertFalse(server.getRooms().contains("dev")); + } + } + } + + @Test + void roomCreationBeyondTheConfiguredCapIsRejected() throws Exception { + int port = 0; + try (ChatServer server = new ChatServer(ChatServerConfig.ofPort(port).withMaxRooms(2))) { + server.start(); + server.awaitStarted(); + port = server.getPort(); + + try (TestClient alice = new TestClient("alice", port)) { + alice.connect(); + alice.joinRoom("dev"); + awaitProtocolEvent(alice, MessageType.ROOM_JOINED, "dev"); + + alice.joinRoom("overflow"); + awaitProtocolEvent(alice, MessageType.ERROR, "Room limit reached"); + + assertFalse(server.getRooms().contains("overflow")); } } } diff --git a/src/main/java/dev/krotname/networkchat/client/ChatClient.java b/src/main/java/dev/krotname/networkchat/client/ChatClient.java index febe1c0..51e866a 100644 --- a/src/main/java/dev/krotname/networkchat/client/ChatClient.java +++ b/src/main/java/dev/krotname/networkchat/client/ChatClient.java @@ -374,6 +374,7 @@ protected void clientMainLoop() throws IOException { case USER_ADDED -> informAboutAddingNewUser(message.data()); case USER_REMOVED -> informAboutDeletingNewUser(message.data()); case ROOM_ADDED -> informAboutRoomAdded(message.room()); + case ROOM_REMOVED -> informAboutRoomRemoved(message.room()); case ROOM_JOINED -> informAboutRoomJoined(message.room()); case ROOM_LEFT -> informAboutRoomLeft(message.room()); case TEXT, ROOM_TEXT, PRIVATE_TEXT -> processIncomingMessage(message); @@ -399,6 +400,10 @@ protected void informAboutRoomAdded(String roomName) { System.out.printf("Комната доступна: %s%n", roomName); } + protected void informAboutRoomRemoved(String roomName) { + System.out.printf("Комната закрыта: %s%n", roomName); + } + protected void informAboutRoomJoined(String roomName) { System.out.printf("Вы вошли в комнату: %s%n", roomName); } @@ -457,7 +462,7 @@ private void validateServerFrame(ChatMessage message) throws IOException { throw new IOException("Server sent an invalid user event"); } } - case ROOM_ADDED, ROOM_JOINED, ROOM_LEFT -> { + case ROOM_ADDED, ROOM_REMOVED, ROOM_JOINED, ROOM_LEFT -> { if (!isValidRoomName(message.room())) { throw new IOException("Server sent an invalid room event"); } diff --git a/src/main/java/dev/krotname/networkchat/client/ClientGuiController.java b/src/main/java/dev/krotname/networkchat/client/ClientGuiController.java index 6bad7a3..efcafaa 100644 --- a/src/main/java/dev/krotname/networkchat/client/ClientGuiController.java +++ b/src/main/java/dev/krotname/networkchat/client/ClientGuiController.java @@ -266,6 +266,19 @@ protected void informAboutRoomAdded(String roomName) { view.refreshRooms(currentRoom); } + @Override + protected void informAboutRoomRemoved(String roomName) { + if (!model.removeRoom(roomName)) { + return; + } + if (roomName.equals(currentRoom)) { + currentRoom = ChatMessage.GENERAL_ROOM; + } + model.addServiceMessage("Комната " + roomName + " закрыта"); + view.refreshRooms(currentRoom); + view.refreshMessages(); + } + @Override protected void informAboutRoomJoined(String roomName) { model.joinRoom(roomName); diff --git a/src/main/java/dev/krotname/networkchat/client/gui/ClientGuiModel.java b/src/main/java/dev/krotname/networkchat/client/gui/ClientGuiModel.java index ef1a5ed..bdd30b0 100644 --- a/src/main/java/dev/krotname/networkchat/client/gui/ClientGuiModel.java +++ b/src/main/java/dev/krotname/networkchat/client/gui/ClientGuiModel.java @@ -162,6 +162,12 @@ public synchronized void addRoom(String roomName) { allRoomNames.add(roomName); } + /** Drops a room reclaimed by the server so the room list does not go stale. */ + public synchronized boolean removeRoom(String roomName) { + joinedRoomNames.remove(roomName); + return allRoomNames.remove(roomName); + } + public synchronized void joinRoom(String roomName) { allRoomNames.add(roomName); joinedRoomNames.add(roomName); diff --git a/src/main/java/dev/krotname/networkchat/network/ChatServer.java b/src/main/java/dev/krotname/networkchat/network/ChatServer.java index 5483cce..c67fd90 100644 --- a/src/main/java/dev/krotname/networkchat/network/ChatServer.java +++ b/src/main/java/dev/krotname/networkchat/network/ChatServer.java @@ -8,7 +8,10 @@ import java.nio.file.Path; import java.time.Duration; import java.time.Instant; +import java.util.ArrayList; import java.util.Collections; +import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; @@ -33,7 +36,6 @@ public final class ChatServer implements AutoCloseable { private static final int MIN_ROOM_NAME_LENGTH = 1; private static final int MAX_ROOM_NAME_LENGTH = 64; private static final int MAX_HANDSHAKE_ATTEMPTS = 5; - static final int MAX_ROOMS = 1_000; static final String ENV_TLS_KEYSTORE_PASSWORD = "NETWORK_CHAT_TLS_KEYSTORE_PASSWORD"; static final String ENV_TLS_KEY_PASSWORD = "NETWORK_CHAT_TLS_KEY_PASSWORD"; @@ -44,6 +46,7 @@ public final class ChatServer implements AutoCloseable { private final Map roles = new ConcurrentHashMap<>(); private final Map> roomMembers = new ConcurrentHashMap<>(); private final Object sessionsMonitor = new Object(); + private final Object roomsMonitor = new Object(); private final AtomicInteger activeClients = new AtomicInteger(); private final ExecutorService clientExecutor; private final ExecutorService acceptorExecutor = @@ -67,7 +70,7 @@ public ChatServer(ChatServerConfig config) { config.accountFile() == null ? AccountStore.disabled() : loadAccountStore(config); roomMembers.put(ChatMessage.GENERAL_ROOM, ConcurrentHashMap.newKeySet()); for (String roomName : historyStore.knownRooms()) { - if (roomMembers.size() >= MAX_ROOMS) { + if (roomMembers.size() >= config.maxRooms()) { break; } roomMembers.putIfAbsent(roomName, ConcurrentHashMap.newKeySet()); @@ -110,6 +113,8 @@ static ChatServerConfig parseConfig(String[] args, Map environme case "--port" -> parsedConfig = parsedConfig.withPort(Integer.parseInt(nextArg(args, ++i, arg))); case "--bind" -> parsedConfig = parsedConfig.withBindAddress(nextArg(args, ++i, arg)); + case "--max-rooms" -> + parsedConfig = parsedConfig.withMaxRooms(Integer.parseInt(nextArg(args, ++i, arg))); case "--history" -> parsedConfig = parsedConfig.withHistory(Path.of(nextArg(args, ++i, arg))); case "--accounts" -> @@ -230,16 +235,17 @@ private void handleClient(Socket socket) { try (socket; ChatConnection connection = new ChatConnection(socket)) { sessionConnection = connection; + ClientLimits limits = new ClientLimits(config.rateLimit()); HandshakeResult handshake = serverHandshake(connection); userName = handshake.userName(); - joinRoom(userName, ChatMessage.GENERAL_ROOM, connection); + joinRoom(userName, ChatMessage.GENERAL_ROOM, connection, limits); socket.setSoTimeout(config.readTimeoutMillis()); sendUsersListToNewClient(connection); sendRoomsListToNewClient(connection); activeClients.incrementAndGet(); active = true; broadcast(ChatMessage.withData(MessageType.USER_ADDED, userName, null), connection); - serverMainLoop(connection, userName); + serverMainLoop(connection, userName, limits); } } catch (IOException | RuntimeException ex) { LOG.log(Level.FINE, "Client session ended", ex); @@ -319,19 +325,29 @@ private void sendUsersListToNewClient(ChatConnection connection) throws IOExcept } private void sendRoomsListToNewClient(ChatConnection connection) throws IOException { - for (String roomName : roomMembers.keySet()) { + List roomNames; + synchronized (roomsMonitor) { + roomNames = List.copyOf(roomMembers.keySet()); + } + for (String roomName : roomNames) { connection.send(ChatMessage.roomAdded(roomName)); } } /** Read messages from one client and broadcast normalized text messages to all clients. */ - private void serverMainLoop(ChatConnection connection, String userName) throws IOException { + private void serverMainLoop(ChatConnection connection, String userName, ClientLimits limits) + throws IOException { while (running.get()) { ChatMessage message = connection.receive(config.readTimeout()); if (!isProtocolSupported(message)) { connection.send(unsupportedProtocolVersionError()); continue; } + if (!limits.frames().tryAcquire()) { + connection.send( + ChatMessage.withData(MessageType.ERROR, "Too many requests, slow down", null)); + continue; + } if (isHealthCommand(message)) { handleHealthCommand(userName, connection); continue; @@ -339,9 +355,16 @@ private void serverMainLoop(ChatConnection connection, String userName) throws I switch (message.type()) { case TEXT, ROOM_TEXT -> handleRoomText(message, userName, connection); case PRIVATE_TEXT -> handlePrivateText(message, userName, connection); - case ROOM_JOIN -> handleRoomJoin(message, userName, connection); + case ROOM_JOIN -> handleRoomJoin(message, userName, connection, limits); case ROOM_LEAVE -> handleRoomLeave(message, userName, connection); - case NAME_ACCEPTED, USER_ADDED, USER_REMOVED, ROOM_ADDED, ROOM_JOINED, ROOM_LEFT, ERROR -> + case NAME_ACCEPTED, + USER_ADDED, + USER_REMOVED, + ROOM_ADDED, + ROOM_REMOVED, + ROOM_JOINED, + ROOM_LEFT, + ERROR -> connection.send( ChatMessage.withData(MessageType.ERROR, "Unsupported client frame", null)); default -> throw new IOException("Unsupported message type: " + message.type()); @@ -436,28 +459,15 @@ private String healthSummary() { config.tls().enabled()); } - private void handleRoomJoin(ChatMessage message, String userName, ChatConnection connection) + private void handleRoomJoin( + ChatMessage message, String userName, ChatConnection connection, ClientLimits limits) throws IOException { String roomName = roomOrGeneral(message.room()); if (!isRoomNameValid(roomName)) { connection.send(ChatMessage.withData(MessageType.ERROR, "Invalid room name", null)); return; } - boolean created = false; - synchronized (roomMembers) { - if (!roomMembers.containsKey(roomName)) { - if (roomMembers.size() >= MAX_ROOMS) { - connection.send(ChatMessage.withData(MessageType.ERROR, "Room limit reached", null)); - return; - } - roomMembers.put(roomName, ConcurrentHashMap.newKeySet()); - created = true; - } - } - if (created) { - broadcast(ChatMessage.roomAdded(roomName), null); - } - joinRoom(userName, roomName, connection); + joinRoom(userName, roomName, connection, limits); } private void handleRoomLeave(ChatMessage message, String userName, ChatConnection connection) @@ -467,21 +477,40 @@ private void handleRoomLeave(ChatMessage message, String userName, ChatConnectio connection.send(ChatMessage.withData(MessageType.ERROR, "Cannot leave general room", null)); return; } - Set members = roomMembers.get(roomName); - if (members != null) { - members.remove(userName); - } + boolean reclaimed = releaseRoom(userName, roomName); connection.send(ChatMessage.roomLeft(roomName)); + if (reclaimed) { + announceRoomRemoved(roomName); + } } - private void joinRoom(String userName, String roomName, ChatConnection connection) + /** + * Admits one user to an existing or newly created room. Creation is bounded by the configured + * room cap and by a per-connection token bucket; rejected requests are answered with an {@code + * ERROR} frame instead of closing the session. + */ + private void joinRoom( + String userName, String roomName, ChatConnection connection, ClientLimits limits) throws IOException { if (sessions.get(userName) != connection) { throw new IOException("Session is no longer active"); } - Set members = - roomMembers.computeIfAbsent(roomName, ignored -> ConcurrentHashMap.newKeySet()); - members.add(userName); + RoomAdmission admission = admitToRoom(userName, roomName, limits); + switch (admission) { + case THROTTLED -> { + connection.send( + ChatMessage.withData(MessageType.ERROR, "Too many new rooms, slow down", null)); + return; + } + case LIMIT_REACHED -> { + connection.send(ChatMessage.withData(MessageType.ERROR, "Room limit reached", null)); + return; + } + case CREATED -> broadcast(ChatMessage.roomAdded(roomName), null); + case JOINED -> { + // The room already existed, so no ROOM_ADDED announcement is needed. + } + } try { connection.send(ChatMessage.roomJoined(roomName)); for (ChatMessage message : @@ -489,11 +518,83 @@ private void joinRoom(String userName, String roomName, ChatConnection connectio connection.send(message); } } catch (IOException ex) { - members.remove(userName); + if (releaseRoom(userName, roomName)) { + announceRoomRemoved(roomName); + } throw ex; } } + /** + * Reserves room membership under a dedicated monitor so the cap is exact and a room cannot be + * reclaimed between a joiner observing it and joining it. No I/O runs while the monitor is held, + * and the monitor is never taken while the sessions monitor is held. + */ + private RoomAdmission admitToRoom(String userName, String roomName, ClientLimits limits) { + synchronized (roomsMonitor) { + Set members = roomMembers.get(roomName); + if (members != null) { + members.add(userName); + return RoomAdmission.JOINED; + } + if (!limits.roomCreations().tryAcquire()) { + return RoomAdmission.THROTTLED; + } + if (roomMembers.size() >= config.maxRooms()) { + return RoomAdmission.LIMIT_REACHED; + } + Set createdMembers = ConcurrentHashMap.newKeySet(); + createdMembers.add(userName); + roomMembers.put(roomName, createdMembers); + return RoomAdmission.CREATED; + } + } + + /** + * Drops one membership and reclaims the room once it is empty, so room names cannot accumulate + * for the lifetime of the process. The general room is never reclaimed. + */ + private boolean releaseRoom(String userName, String roomName) { + synchronized (roomsMonitor) { + Set members = roomMembers.get(roomName); + if (members == null || !members.remove(userName) || !members.isEmpty()) { + return false; + } + if (ChatMessage.GENERAL_ROOM.equals(roomName)) { + return false; + } + roomMembers.remove(roomName); + return true; + } + } + + /** Drops every membership of one user and returns the rooms reclaimed by that disconnect. */ + private List releaseAllRooms(String userName) { + List reclaimedRooms = new ArrayList<>(); + synchronized (roomsMonitor) { + Iterator>> rooms = roomMembers.entrySet().iterator(); + while (rooms.hasNext()) { + Map.Entry> room = rooms.next(); + if (!room.getValue().remove(userName) + || !room.getValue().isEmpty() + || ChatMessage.GENERAL_ROOM.equals(room.getKey())) { + continue; + } + rooms.remove(); + reclaimedRooms.add(room.getKey()); + } + } + return reclaimedRooms; + } + + private void announceRoomRemoved(String roomName) { + try { + broadcast(ChatMessage.roomRemoved(roomName), null); + } catch (IOException ex) { + LOG.log(Level.FINE, "Unable to broadcast ROOM_REMOVED", ex); + } + } + private boolean isRoomMember(String userName, String roomName) { Set members = roomMembers.get(roomName); return members != null && members.contains(userName); @@ -557,23 +658,28 @@ private void broadcast(ChatMessage message, ChatConnection exceptConnection) thr } } + /** + * Drops one session and reclaims the rooms it emptied. Announcements run outside the sessions + * monitor so the rooms monitor is never nested inside it. + */ boolean removeSession(String userName, ChatConnection expectedConnection) { synchronized (sessionsMonitor) { if (expectedConnection == null || !sessions.remove(userName, expectedConnection)) { return false; } roles.remove(userName); - for (Set members : roomMembers.values()) { - members.remove(userName); - } closeConnection(expectedConnection); - try { - broadcast(ChatMessage.withData(MessageType.USER_REMOVED, userName, null), null); - } catch (IOException ex) { - LOG.log(Level.FINE, "Unable to broadcast USER_REMOVED", ex); - } - return true; } + List reclaimedRooms = releaseAllRooms(userName); + try { + broadcast(ChatMessage.withData(MessageType.USER_REMOVED, userName, null), null); + } catch (IOException ex) { + LOG.log(Level.FINE, "Unable to broadcast USER_REMOVED", ex); + } + for (String reclaimedRoom : reclaimedRooms) { + announceRoomRemoved(reclaimedRoom); + } + return true; } @Override @@ -662,6 +768,8 @@ private void closeAllSessions() { closingSessions = Map.copyOf(sessions); sessions.clear(); roles.clear(); + } + synchronized (roomsMonitor) { for (Set members : roomMembers.values()) { members.removeAll(closingSessions.keySet()); } @@ -680,4 +788,19 @@ private static void closeConnection(ChatConnection connection) { } private record HandshakeResult(String userName, UserRole role) {} + + /** Per-connection throttles created once for every accepted client. */ + private record ClientLimits(TokenBucket frames, TokenBucket roomCreations) { + ClientLimits(RateLimitConfig rateLimit) { + this(rateLimit.newFrameBucket(), rateLimit.newRoomCreationBucket()); + } + } + + /** Result of one room admission attempt. */ + private enum RoomAdmission { + JOINED, + CREATED, + LIMIT_REACHED, + THROTTLED + } } diff --git a/src/main/java/dev/krotname/networkchat/network/ChatServerConfig.java b/src/main/java/dev/krotname/networkchat/network/ChatServerConfig.java index 4c2a4ff..9bfe007 100644 --- a/src/main/java/dev/krotname/networkchat/network/ChatServerConfig.java +++ b/src/main/java/dev/krotname/networkchat/network/ChatServerConfig.java @@ -9,17 +9,21 @@ public record ChatServerConfig( int port, String bindAddress, int maxClients, + int maxRooms, Duration handshakeTimeout, Duration readTimeout, Path historyFile, int historyLimit, int historyReplayLimit, Path accountFile, - TlsServerConfig tls) { + TlsServerConfig tls, + RateLimitConfig rateLimit) { public static final int DEFAULT_PORT = 1500; public static final String DEFAULT_BIND_ADDRESS = "127.0.0.1"; public static final int DEFAULT_MAX_CLIENTS = 100; public static final int MAX_CLIENTS = 10_000; + public static final int DEFAULT_MAX_ROOMS = 256; + public static final int MAX_ROOMS = 100_000; public static final Duration DEFAULT_HANDSHAKE_TIMEOUT = Duration.ofSeconds(10); public static final Duration DEFAULT_READ_TIMEOUT = Duration.ofMinutes(5); public static final int DEFAULT_HISTORY_LIMIT = 10_000; @@ -33,13 +37,15 @@ public ChatServerConfig( port, DEFAULT_BIND_ADDRESS, maxClients, + DEFAULT_MAX_ROOMS, handshakeTimeout, readTimeout, null, DEFAULT_HISTORY_LIMIT, DEFAULT_HISTORY_REPLAY_LIMIT, null, - TlsServerConfig.disabled()); + TlsServerConfig.disabled(), + RateLimitConfig.defaultLimits()); } public ChatServerConfig { @@ -56,6 +62,9 @@ public ChatServerConfig( if (maxClients < 1 || maxClients > MAX_CLIENTS) { throw new IllegalArgumentException("Max clients must be in range 1.." + MAX_CLIENTS); } + if (maxRooms < 1 || maxRooms > MAX_ROOMS) { + throw new IllegalArgumentException("Max rooms must be in range 1.." + MAX_ROOMS); + } Objects.requireNonNull(handshakeTimeout, "handshakeTimeout"); Objects.requireNonNull(readTimeout, "readTimeout"); if (handshakeTimeout.isNegative() || handshakeTimeout.isZero()) { @@ -74,6 +83,7 @@ public ChatServerConfig( "History replay limit must be between 0 and the configured limits"); } Objects.requireNonNull(tls, "tls"); + Objects.requireNonNull(rateLimit, "rateLimit"); validateSocketTimeout(handshakeTimeout, "handshakeTimeout"); validateSocketTimeout(readTimeout, "readTimeout"); } @@ -92,13 +102,15 @@ public ChatServerConfig withPort(int newPort) { newPort, bindAddress, maxClients, + maxRooms, handshakeTimeout, readTimeout, historyFile, historyLimit, historyReplayLimit, accountFile, - tls); + tls, + rateLimit); } public ChatServerConfig withBindAddress(String newBindAddress) { @@ -106,13 +118,31 @@ public ChatServerConfig withBindAddress(String newBindAddress) { port, newBindAddress, maxClients, + maxRooms, + handshakeTimeout, + readTimeout, + historyFile, + historyLimit, + historyReplayLimit, + accountFile, + tls, + rateLimit); + } + + public ChatServerConfig withMaxRooms(int newMaxRooms) { + return new ChatServerConfig( + port, + bindAddress, + maxClients, + newMaxRooms, handshakeTimeout, readTimeout, historyFile, historyLimit, historyReplayLimit, accountFile, - tls); + tls, + rateLimit); } public ChatServerConfig withHistory(Path newHistoryFile) { @@ -120,13 +150,15 @@ public ChatServerConfig withHistory(Path newHistoryFile) { port, bindAddress, maxClients, + maxRooms, handshakeTimeout, readTimeout, newHistoryFile, historyLimit, historyReplayLimit, accountFile, - tls); + tls, + rateLimit); } public ChatServerConfig withHistory(Path newHistoryFile, int newLimit, int newReplayLimit) { @@ -134,13 +166,15 @@ public ChatServerConfig withHistory(Path newHistoryFile, int newLimit, int newRe port, bindAddress, maxClients, + maxRooms, handshakeTimeout, readTimeout, newHistoryFile, newLimit, newReplayLimit, accountFile, - tls); + tls, + rateLimit); } public ChatServerConfig withAccounts(Path newAccountFile) { @@ -148,13 +182,15 @@ public ChatServerConfig withAccounts(Path newAccountFile) { port, bindAddress, maxClients, + maxRooms, handshakeTimeout, readTimeout, historyFile, historyLimit, historyReplayLimit, newAccountFile, - tls); + tls, + rateLimit); } public ChatServerConfig withTls(TlsServerConfig newTls) { @@ -162,13 +198,31 @@ public ChatServerConfig withTls(TlsServerConfig newTls) { port, bindAddress, maxClients, + maxRooms, + handshakeTimeout, + readTimeout, + historyFile, + historyLimit, + historyReplayLimit, + accountFile, + newTls, + rateLimit); + } + + public ChatServerConfig withRateLimit(RateLimitConfig newRateLimit) { + return new ChatServerConfig( + port, + bindAddress, + maxClients, + maxRooms, handshakeTimeout, readTimeout, historyFile, historyLimit, historyReplayLimit, accountFile, - newTls); + tls, + newRateLimit); } int handshakeTimeoutMillis() { diff --git a/src/main/java/dev/krotname/networkchat/network/RateLimitConfig.java b/src/main/java/dev/krotname/networkchat/network/RateLimitConfig.java new file mode 100644 index 0000000..b12096d --- /dev/null +++ b/src/main/java/dev/krotname/networkchat/network/RateLimitConfig.java @@ -0,0 +1,51 @@ +package dev.krotname.networkchat.network; + +/** + * Per-connection token-bucket limits. + * + *

Limits are applied per client connection and answered with an {@code ERROR} frame instead of a + * disconnect, so a misbehaving client is slowed down while legitimate bursts still pass. + */ +public record RateLimitConfig( + int frameBurst, double framesPerSecond, int roomCreationBurst, double roomCreationsPerSecond) { + public static final int DEFAULT_FRAME_BURST = 60; + public static final double DEFAULT_FRAMES_PER_SECOND = 20; + public static final int DEFAULT_ROOM_CREATION_BURST = 10; + public static final double DEFAULT_ROOM_CREATIONS_PER_SECOND = 0.5; + public static final int MAX_BURST = 100_000; + + public RateLimitConfig { + validateBurst(frameBurst, "frameBurst"); + validateBurst(roomCreationBurst, "roomCreationBurst"); + validateRate(framesPerSecond, "framesPerSecond"); + validateRate(roomCreationsPerSecond, "roomCreationsPerSecond"); + } + + public static RateLimitConfig defaultLimits() { + return new RateLimitConfig( + DEFAULT_FRAME_BURST, + DEFAULT_FRAMES_PER_SECOND, + DEFAULT_ROOM_CREATION_BURST, + DEFAULT_ROOM_CREATIONS_PER_SECOND); + } + + TokenBucket newFrameBucket() { + return new TokenBucket(frameBurst, framesPerSecond); + } + + TokenBucket newRoomCreationBucket() { + return new TokenBucket(roomCreationBurst, roomCreationsPerSecond); + } + + private static void validateBurst(int burst, String fieldName) { + if (burst < 1 || burst > MAX_BURST) { + throw new IllegalArgumentException(fieldName + " must be in range 1.." + MAX_BURST); + } + } + + private static void validateRate(double rate, String fieldName) { + if (!Double.isFinite(rate) || rate <= 0) { + throw new IllegalArgumentException(fieldName + " must be a positive finite rate"); + } + } +} diff --git a/src/main/java/dev/krotname/networkchat/network/TokenBucket.java b/src/main/java/dev/krotname/networkchat/network/TokenBucket.java new file mode 100644 index 0000000..706aa7c --- /dev/null +++ b/src/main/java/dev/krotname/networkchat/network/TokenBucket.java @@ -0,0 +1,54 @@ +package dev.krotname.networkchat.network; + +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; + +/** + * Minimal token bucket used to throttle a single client connection. + * + *

The bucket starts full so short legitimate bursts are never delayed, and it refills lazily + * from a monotonic clock so no background timer is required. + */ +final class TokenBucket { + private static final double NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1); + + private final double capacity; + private final double tokensPerNano; + private final LongSupplier nanoClock; + private double tokens; + private long lastRefillNanos; + + TokenBucket(int capacity, double permitsPerSecond) { + this(capacity, permitsPerSecond, System::nanoTime); + } + + TokenBucket(int capacity, double permitsPerSecond, LongSupplier nanoClock) { + if (capacity < 1) { + throw new IllegalArgumentException("Token bucket capacity must be positive"); + } + if (!Double.isFinite(permitsPerSecond) || permitsPerSecond <= 0) { + throw new IllegalArgumentException("Token bucket rate must be positive"); + } + this.capacity = capacity; + this.tokensPerNano = permitsPerSecond / NANOS_PER_SECOND; + this.nanoClock = Objects.requireNonNull(nanoClock, "nanoClock"); + this.tokens = capacity; + this.lastRefillNanos = nanoClock.getAsLong(); + } + + /** Consumes one permit and reports whether the caller may proceed. */ + synchronized boolean tryAcquire() { + long now = nanoClock.getAsLong(); + long elapsedNanos = now - lastRefillNanos; + if (elapsedNanos > 0) { + tokens = Math.min(capacity, tokens + elapsedNanos * tokensPerNano); + lastRefillNanos = now; + } + if (tokens < 1) { + return false; + } + tokens -= 1; + return true; + } +} diff --git a/src/main/java/dev/krotname/networkchat/protocol/ChatMessage.java b/src/main/java/dev/krotname/networkchat/protocol/ChatMessage.java index 1f9d063..32b88ba 100644 --- a/src/main/java/dev/krotname/networkchat/protocol/ChatMessage.java +++ b/src/main/java/dev/krotname/networkchat/protocol/ChatMessage.java @@ -108,6 +108,10 @@ public static ChatMessage roomAdded(String room) { return withRoom(MessageType.ROOM_ADDED, room); } + public static ChatMessage roomRemoved(String room) { + return withRoom(MessageType.ROOM_REMOVED, room); + } + public static ChatMessage roomJoined(String room) { return withRoom(MessageType.ROOM_JOINED, room); } diff --git a/src/main/java/dev/krotname/networkchat/protocol/MessageType.java b/src/main/java/dev/krotname/networkchat/protocol/MessageType.java index 1e888c0..8377379 100644 --- a/src/main/java/dev/krotname/networkchat/protocol/MessageType.java +++ b/src/main/java/dev/krotname/networkchat/protocol/MessageType.java @@ -13,6 +13,7 @@ public enum MessageType { ROOM_JOIN, ROOM_LEAVE, ROOM_ADDED, + ROOM_REMOVED, ROOM_JOINED, ROOM_LEFT, ERROR diff --git a/src/test/java/dev/krotname/networkchat/ClientGuiModelTest.java b/src/test/java/dev/krotname/networkchat/ClientGuiModelTest.java index a71d388..4bcf7f3 100644 --- a/src/test/java/dev/krotname/networkchat/ClientGuiModelTest.java +++ b/src/test/java/dev/krotname/networkchat/ClientGuiModelTest.java @@ -53,6 +53,20 @@ void tracksRoomsAndJoinedRooms() { assertTrue(model.getJoinedRoomNames().isEmpty()); } + @Test + void dropsRoomsReclaimedByTheServer() { + ClientGuiModel model = new ClientGuiModel(); + model.joinRoom("dev"); + model.addRoom("ops"); + + assertTrue(model.removeRoom("dev")); + assertFalse(model.removeRoom("dev")); + + assertFalse(model.getAllRoomNames().contains("dev")); + assertFalse(model.getJoinedRoomNames().contains("dev")); + assertTrue(model.getAllRoomNames().contains("ops")); + } + @Test void deduplicatesTimelineByMessageId() { ClientGuiModel model = new ClientGuiModel(); diff --git a/src/test/java/dev/krotname/networkchat/network/ChatServerConcurrencyTest.java b/src/test/java/dev/krotname/networkchat/network/ChatServerConcurrencyTest.java index a97a21f..91dd8e5 100644 --- a/src/test/java/dev/krotname/networkchat/network/ChatServerConcurrencyTest.java +++ b/src/test/java/dev/krotname/networkchat/network/ChatServerConcurrencyTest.java @@ -15,9 +15,8 @@ import java.net.Socket; import java.time.Duration; import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import org.awaitility.Awaitility; import org.junit.jupiter.api.Test; class ChatServerConcurrencyTest { @@ -75,23 +74,118 @@ void closingServerClearsSessionsAndClosesTheirSockets() throws Exception { @Test void roomCreationStopsAtTheConfiguredServerLimit() throws Exception { - try (ChatServer server = new ChatServer(0)) { + try (ChatServer server = new ChatServer(ChatServerConfig.ofPort(0).withMaxRooms(2))) { server.start(); try (ClientSession alice = connect(server, "alice")) { - Map> rooms = rooms(server); - for (int index = rooms.size(); index < ChatServer.MAX_ROOMS; index++) { - rooms.put("existing_" + index, ConcurrentHashMap.newKeySet()); - } + alice.connection().send(ChatMessage.roomJoin("first")); + receiveUntilRoom(alice, "first"); alice.connection().send(ChatMessage.roomJoin("overflow")); ChatMessage response = receiveUntil(alice, MessageType.ERROR); - assertTrue(response.data().contains("limit")); + assertEquals("Room limit reached", response.data()); assertFalse(server.getRooms().contains("overflow")); + assertEquals(2, server.getRooms().size()); + } + } + } + + @Test + void roomCreationIsThrottledPerConnection() throws Exception { + ChatServerConfig config = + ChatServerConfig.ofPort(0).withRateLimit(new RateLimitConfig(60, 20, 1, 0.001)); + try (ChatServer server = new ChatServer(config)) { + server.start(); + try (ClientSession alice = connect(server, "alice")) { + alice.connection().send(ChatMessage.roomJoin("first")); + receiveUntilRoom(alice, "first"); + + alice.connection().send(ChatMessage.roomJoin("second")); + ChatMessage response = receiveUntil(alice, MessageType.ERROR); + + assertEquals("Too many new rooms, slow down", response.data()); + assertFalse(server.getRooms().contains("second")); + assertTrue(server.getConnectedUsers().contains("alice")); } } } + @Test + void inboundFramesAreThrottledPerConnection() throws Exception { + ChatServerConfig config = + ChatServerConfig.ofPort(0).withRateLimit(new RateLimitConfig(1, 0.001, 10, 0.5)); + try (ChatServer server = new ChatServer(config)) { + server.start(); + try (ClientSession alice = connect(server, "alice")) { + alice.connection().send(ChatMessage.roomJoin("first")); + receiveUntilRoom(alice, "first"); + + alice.connection().send(ChatMessage.text("throttled", "alice")); + ChatMessage response = receiveUntil(alice, MessageType.ERROR); + + assertEquals("Too many requests, slow down", response.data()); + assertTrue(server.getConnectedUsers().contains("alice")); + } + } + } + + @Test + void emptyRoomIsReclaimedWhenTheLastMemberLeaves() throws Exception { + try (ChatServer server = new ChatServer(0)) { + server.start(); + try (ClientSession alice = connect(server, "alice"); + ClientSession observer = connect(server, "observer")) { + alice.connection().send(ChatMessage.roomJoin("dev")); + receiveUntilRoom(alice, "dev"); + assertTrue(server.getRooms().contains("dev")); + + alice.connection().send(ChatMessage.roomLeave("dev")); + ChatMessage removed = receiveUntilRoomEvent(observer, MessageType.ROOM_REMOVED, "dev"); + + assertEquals("dev", removed.room()); + assertFalse(server.getRooms().contains("dev")); + assertTrue(server.getRooms().contains(ChatMessage.GENERAL_ROOM)); + } + } + } + + @Test + void emptyRoomIsReclaimedWhenTheLastMemberDisconnects() throws Exception { + try (ChatServer server = new ChatServer(0)) { + server.start(); + try (ClientSession observer = connect(server, "observer")) { + try (ClientSession alice = connect(server, "alice")) { + alice.connection().send(ChatMessage.roomJoin("dev")); + receiveUntilRoom(alice, "dev"); + assertTrue(server.getRooms().contains("dev")); + } + + ChatMessage removed = receiveUntilRoomEvent(observer, MessageType.ROOM_REMOVED, "dev"); + + assertEquals("dev", removed.room()); + assertFalse(server.getRooms().contains("dev")); + } + } + } + + @Test + void generalRoomIsNeverReclaimed() throws Exception { + try (ChatServer server = new ChatServer(0)) { + server.start(); + try (ClientSession alice = connect(server, "alice")) { + alice.connection().send(ChatMessage.roomLeave(ChatMessage.GENERAL_ROOM)); + ChatMessage response = receiveUntil(alice, MessageType.ERROR); + + assertEquals("Cannot leave general room", response.data()); + } + + Awaitility.await("wait for the disconnect to be processed") + .atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> assertTrue(server.getConnectedUsers().isEmpty())); + assertTrue(server.getRooms().contains(ChatMessage.GENERAL_ROOM)); + } + } + @Test void privateHealthTextIsDeliveredInsteadOfBeingTreatedAsAnAdminCommand() throws Exception { try (ChatServer server = new ChatServer(0)) { @@ -162,10 +256,15 @@ private static ChatMessage receiveUntil(ClientSession session, MessageType expec private static ChatMessage receiveUntilRoom(ClientSession session, String expectedRoom) throws Exception { + return receiveUntilRoomEvent(session, MessageType.ROOM_JOINED, expectedRoom); + } + + private static ChatMessage receiveUntilRoomEvent( + ClientSession session, MessageType expectedType, String expectedRoom) throws Exception { ChatMessage response; do { response = session.connection().receive(Duration.ofSeconds(2)); - } while (response.type() != MessageType.ROOM_JOINED || !expectedRoom.equals(response.room())); + } while (response.type() != expectedType || !expectedRoom.equals(response.room())); return response; } @@ -176,13 +275,6 @@ private static Map sessions(ChatServer server) throws Ex return (Map) field.get(server); } - @SuppressWarnings("unchecked") - private static Map> rooms(ChatServer server) throws Exception { - Field field = ChatServer.class.getDeclaredField("roomMembers"); - field.setAccessible(true); - return (Map>) field.get(server); - } - private record ClientSession(Socket socket, ChatConnection connection) implements Closeable { @Override public void close() throws IOException { diff --git a/src/test/java/dev/krotname/networkchat/network/ChatServerConfigTest.java b/src/test/java/dev/krotname/networkchat/network/ChatServerConfigTest.java index e10aca5..3b5a563 100644 --- a/src/test/java/dev/krotname/networkchat/network/ChatServerConfigTest.java +++ b/src/test/java/dev/krotname/networkchat/network/ChatServerConfigTest.java @@ -2,6 +2,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import java.net.InetAddress; @@ -104,6 +105,40 @@ void supportsAccountsHistoryAndTlsConfig() { assertEquals(tls, config.tls()); } + @Test + void exposesRoomCapAndPerConnectionRateLimits() { + ChatServerConfig defaults = ChatServerConfig.defaultConfig(); + + assertEquals(ChatServerConfig.DEFAULT_MAX_ROOMS, defaults.maxRooms()); + assertEquals(RateLimitConfig.defaultLimits(), defaults.rateLimit()); + + ChatServerConfig tightened = + defaults.withMaxRooms(8).withRateLimit(new RateLimitConfig(5, 1, 2, 0.5)); + + assertEquals(8, tightened.maxRooms()); + assertEquals(2, tightened.rateLimit().roomCreationBurst()); + } + + @Test + void rejectsInvalidRoomCapAndRateLimits() { + ChatServerConfig defaults = ChatServerConfig.defaultConfig(); + + assertThrows(IllegalArgumentException.class, () -> assertNotNull(defaults.withMaxRooms(0))); + assertThrows( + IllegalArgumentException.class, + () -> assertNotNull(defaults.withMaxRooms(ChatServerConfig.MAX_ROOMS + 1))); + assertThrows(NullPointerException.class, () -> assertNotNull(defaults.withRateLimit(null))); + assertThrows(IllegalArgumentException.class, () -> new RateLimitConfig(0, 1, 1, 1)); + assertThrows(IllegalArgumentException.class, () -> new RateLimitConfig(1, 1, 0, 1)); + assertThrows(IllegalArgumentException.class, () -> new RateLimitConfig(1, 0, 1, 1)); + assertThrows( + IllegalArgumentException.class, + () -> new RateLimitConfig(1, 1, 1, Double.POSITIVE_INFINITY)); + assertThrows( + IllegalArgumentException.class, + () -> new RateLimitConfig(RateLimitConfig.MAX_BURST + 1, 1, 1, 1)); + } + @Test void rejectsIncompleteTlsConfig() { assertThrows(IllegalArgumentException.class, () -> TlsServerConfig.enabled(null, "changeit")); @@ -115,7 +150,12 @@ void rejectsIncompleteTlsConfig() { void parsesServerOptionsAndReadsTlsSecretsOnlyFromEnvironment() { ChatServerConfig config = ChatServer.parseConfig( - new String[] {"--port", "1600", "--bind", "0.0.0.0", "--tls-keystore", "chat.p12"}, + new String[] { + "--port", "1600", + "--bind", "0.0.0.0", + "--max-rooms", "16", + "--tls-keystore", "chat.p12" + }, Map.of( ChatServer.ENV_TLS_KEYSTORE_PASSWORD, "store-secret", @@ -124,6 +164,7 @@ void parsesServerOptionsAndReadsTlsSecretsOnlyFromEnvironment() { assertEquals(1600, config.port()); assertEquals("0.0.0.0", config.bindAddress()); + assertEquals(16, config.maxRooms()); assertEquals("store-secret", config.tls().keyStorePassword()); assertEquals("key-secret", config.tls().keyPassword()); assertThrows( diff --git a/src/test/java/dev/krotname/networkchat/network/TokenBucketTest.java b/src/test/java/dev/krotname/networkchat/network/TokenBucketTest.java new file mode 100644 index 0000000..c20b857 --- /dev/null +++ b/src/test/java/dev/krotname/networkchat/network/TokenBucketTest.java @@ -0,0 +1,47 @@ +package dev.krotname.networkchat.network; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Test; + +class TokenBucketTest { + + @Test + void throttlesAfterTheBurstAndRefillsOverTime() { + AtomicLong clock = new AtomicLong(); + TokenBucket bucket = new TokenBucket(2, 1, clock::get); + + assertTrue(bucket.tryAcquire()); + assertTrue(bucket.tryAcquire()); + assertFalse(bucket.tryAcquire()); + + clock.addAndGet(TimeUnit.SECONDS.toNanos(1)); + + assertTrue(bucket.tryAcquire()); + assertFalse(bucket.tryAcquire()); + } + + @Test + void refillNeverExceedsTheConfiguredBurst() { + AtomicLong clock = new AtomicLong(); + TokenBucket bucket = new TokenBucket(2, 1, clock::get); + + clock.addAndGet(TimeUnit.HOURS.toNanos(1)); + + assertTrue(bucket.tryAcquire()); + assertTrue(bucket.tryAcquire()); + assertFalse(bucket.tryAcquire()); + } + + @Test + void rejectsInvalidBucketSettings() { + assertThrows(IllegalArgumentException.class, () -> new TokenBucket(0, 1)); + assertThrows(IllegalArgumentException.class, () -> new TokenBucket(1, 0)); + assertThrows(IllegalArgumentException.class, () -> new TokenBucket(1, Double.NaN)); + assertThrows(NullPointerException.class, () -> new TokenBucket(1, 1, null)); + } +}