diff --git a/resources/lang/en.json b/resources/lang/en.json index 6c89617f8b..09f5ce3dcc 100644 --- a/resources/lang/en.json +++ b/resources/lang/en.json @@ -890,7 +890,8 @@ "admin": "Kicked by an admin", "duplicate_session": "Kicked from game (you may have been playing on another tab)", "host_left": "The host has left the lobby.", - "lobby_creator": "Kicked by lobby creator" + "lobby_creator": "Kicked by lobby creator", + "too_much_data": "Kicked for sending too much data" }, "lang": { "en": "English", diff --git a/src/client/PacedSender.ts b/src/client/PacedSender.ts new file mode 100644 index 0000000000..5935673fe6 --- /dev/null +++ b/src/client/PacedSender.ts @@ -0,0 +1,37 @@ +/** + * Sends queued work at a fixed minimum spacing. + * + * The server drops intents past its per-second budget, so a burst of batches + * has to be spread out rather than emitted all at once. + */ +export class PacedSender { + private readonly queue: (() => boolean)[] = []; + private timer: ReturnType | null = null; + + constructor(private readonly intervalMs: number) {} + + push(send: () => boolean): void { + this.queue.push(send); + if (this.timer === null) this.next(); + } + + clear(): void { + this.queue.length = 0; + if (this.timer !== null) { + clearTimeout(this.timer); + this.timer = null; + } + } + + private next(): void { + const send = this.queue[0]; + if (send === undefined) { + this.timer = null; + return; + } + // Keep the item queued if it could not be sent, so a reconnect does not + // lose it. leaveGame() clears the queue when the game is over. + if (send()) this.queue.shift(); + this.timer = setTimeout(() => this.next(), this.intervalMs); + } +} diff --git a/src/client/Transport.ts b/src/client/Transport.ts index 2ad958ab11..ace1374245 100644 --- a/src/client/Transport.ts +++ b/src/client/Transport.ts @@ -12,6 +12,7 @@ import { import { TileRef } from "../core/game/GameMap"; import { AllPlayersStats, + batchMoveWarshipUnitIds, ClientHashMessage, ClientIntentMessage, ClientJoinMessage, @@ -32,9 +33,15 @@ import { getPlayToken } from "./Auth"; import { LobbyConfig } from "./ClientGameRunner"; import { showInGameAlert } from "./InGameModal"; import { LocalServer } from "./LocalServer"; +import { PacedSender } from "./PacedSender"; import { translateText } from "./Utils"; import { PlayerView } from "./view"; +// Spacing between batched move_warship intents. The server allows 10 intents +// per second across all types; this stays clear of that with room for the +// player's other actions. +const WARSHIP_BATCH_INTERVAL_MS = 150; + export class PauseGameIntentEvent implements GameEvent { constructor(public readonly paused: boolean) {} } @@ -202,6 +209,10 @@ export class Transport { private onmessage: (msg: ServerMessage) => void; private pingInterval: number | null = null; + private readonly warshipBatches: PacedSender; + // The server discards intents until it has processed our join, and an open + // socket says nothing about that. Its first message does. + private joined = false; public readonly isLocal: boolean; constructor( @@ -213,6 +224,10 @@ export class Transport { this.isLocal = lobbyConfig.gameRecord !== undefined || lobbyConfig.gameStartInfo?.config.gameType === GameType.Singleplayer; + // The local server has no rate limiter, so only remote games need spacing. + this.warshipBatches = new PacedSender( + this.isLocal ? 0 : WARSHIP_BATCH_INTERVAL_MS, + ); this.eventBus.on(SendAllianceRequestIntentEvent, (e) => this.onSendAllianceRequest(e), @@ -351,6 +366,7 @@ export class Transport { // the desktop app://openfront origin), not window.location.host. const workerPath = ClientEnv.workerPath(this.lobbyConfig.gameID); this.socket = new WebSocket(`${ClientEnv.serverWsBase()}/${workerPath}`); + this.joined = false; this.onconnect = onconnect; this.onmessage = onmessage; this.socket.onopen = () => { @@ -371,6 +387,7 @@ export class Transport { onconnect(); }; this.socket.onmessage = (event: MessageEvent) => { + this.joined = true; try { const parsed = JSON.parse(event.data); const result = ServerMessageSchema.safeParse(parsed); @@ -441,6 +458,7 @@ export class Transport { } leaveGame() { + this.warshipBatches.clear(); if (this.isLocal) { this.localServer.endGame(); return; @@ -649,11 +667,15 @@ export class Transport { } private onMoveWarshipEvent(event: MoveWarshipIntentEvent) { - this.sendIntent({ - type: "move_warship", - unitIds: event.unitIds, - tile: event.tile, - }); + for (const unitIds of batchMoveWarshipUnitIds(event.unitIds, event.tile)) { + this.warshipBatches.push(() => + this.sendIntent({ + type: "move_warship", + unitIds, + tile: event.tile, + }), + ); + } } private onSendDeleteUnitIntent(event: SendDeleteUnitIntentEvent) { @@ -681,20 +703,24 @@ export class Transport { this.sendIntent({ type: "toggle_game_start_timer" }); } - private sendIntent(intent: Intent) { - if (this.isLocal || this.socket?.readyState === WebSocket.OPEN) { + private sendIntent(intent: Intent): boolean { + if ( + this.isLocal || + (this.joined && this.socket?.readyState === WebSocket.OPEN) + ) { const msg = { type: "intent", intent: intent, } satisfies ClientIntentMessage; this.sendMsg(msg); - } else { - console.log( - "WebSocket is not open. Current state:", - this.socket?.readyState, - ); - console.log("attempting reconnect"); + return true; } + console.log( + "WebSocket is not open. Current state:", + this.socket?.readyState, + ); + console.log("attempting reconnect"); + return false; } private sendMsg(msg: ClientMessage) { diff --git a/src/core/Schemas.ts b/src/core/Schemas.ts index f47dfa1961..3fd0319e6b 100644 --- a/src/core/Schemas.ts +++ b/src/core/Schemas.ts @@ -534,10 +534,35 @@ export const CancelBoatIntentSchema = z.object({ export const MoveWarshipIntentSchema = z.object({ type: z.literal("move_warship"), - unitIds: z.array(z.number().int()).nonempty(), + unitIds: z.array(z.number().int()).nonempty().max(1000), tile: z.number(), }); +// Client messages larger than this get the client kicked by ClientMsgRateLimiter. +export const MAX_INTENT_SIZE = 2000; + +export function batchMoveWarshipUnitIds( + unitIds: readonly number[], + tile: number, +): number[][] { + const overhead = JSON.stringify({ + type: "intent", + intent: { type: "move_warship", unitIds: [], tile }, + } satisfies ClientIntentMessage).length; + const batches: number[][] = []; + let size = overhead; + for (const unitId of unitIds) { + const unitIdSize = String(unitId).length + 1; + if (batches.length === 0 || size + unitIdSize > MAX_INTENT_SIZE) { + batches.push([]); + size = overhead; + } + batches[batches.length - 1].push(unitId); + size += unitIdSize; + } + return batches; +} + export const DeleteUnitIntentSchema = z.object({ type: z.literal("delete_unit"), unitId: z.number(), diff --git a/src/server/ClientMsgRateLimiter.ts b/src/server/ClientMsgRateLimiter.ts index 96b408092f..6c0dbee42e 100644 --- a/src/server/ClientMsgRateLimiter.ts +++ b/src/server/ClientMsgRateLimiter.ts @@ -1,9 +1,8 @@ import { RateLimiter } from "limiter"; -import { ClientID } from "../core/Schemas"; +import { ClientID, MAX_INTENT_SIZE } from "../core/Schemas"; const INTENTS_PER_SECOND = 10; const INTENTS_PER_MINUTE = 150; -const MAX_INTENT_SIZE = 2000; const TOTAL_BYTES = 5 * 1024 * 1024; // 5MB per client export type RateLimitResult = "ok" | "limit" | "kick"; diff --git a/tests/MoveWarshipIntentBatching.test.ts b/tests/MoveWarshipIntentBatching.test.ts new file mode 100644 index 0000000000..12b4e7c91e --- /dev/null +++ b/tests/MoveWarshipIntentBatching.test.ts @@ -0,0 +1,56 @@ +import { + batchMoveWarshipUnitIds, + ClientIntentMessage, + ClientMessageSchema, + MAX_INTENT_SIZE, +} from "../src/core/Schemas"; +import { replacer } from "../src/core/Util"; + +const TILE = 250_000; + +function frame(unitIds: number[], tile: number): string { + return JSON.stringify( + { + type: "intent", + intent: { type: "move_warship", unitIds, tile }, + } satisfies ClientIntentMessage, + replacer, + ); +} + +describe("batchMoveWarshipUnitIds", () => { + test("sends a small fleet as a single intent", () => { + const unitIds = [10000, 10003, 10006]; + expect(batchMoveWarshipUnitIds(unitIds, TILE)).toEqual([unitIds]); + }); + + test("returns no batches for an empty selection", () => { + expect(batchMoveWarshipUnitIds([], TILE)).toEqual([]); + }); + + test.each([1, 400, 4000])( + "keeps every batch under the server cap (%i warships)", + (count) => { + const unitIds = Array.from({ length: count }, (_, i) => 900_000 + i * 3); + const batches = batchMoveWarshipUnitIds(unitIds, TILE); + + for (const batch of batches) { + expect(batch.length).toBeGreaterThan(0); + expect( + Buffer.byteLength(frame(batch, TILE), "utf8"), + ).toBeLessThanOrEqual(MAX_INTENT_SIZE); + } + expect(batches.flat()).toEqual(unitIds); + }, + ); + + test("every batch is a valid client intent message", () => { + const unitIds = Array.from({ length: 450 }, (_, i) => 10_000 + i * 3); + + for (const batch of batchMoveWarshipUnitIds(unitIds, TILE)) { + expect( + ClientMessageSchema.safeParse(JSON.parse(frame(batch, TILE))).success, + ).toBe(true); + } + }); +}); diff --git a/tests/client/PacedSender.test.ts b/tests/client/PacedSender.test.ts new file mode 100644 index 0000000000..9b1423b3a2 --- /dev/null +++ b/tests/client/PacedSender.test.ts @@ -0,0 +1,132 @@ +import { PacedSender } from "../../src/client/PacedSender"; + +const INTERVAL = 150; + +describe("PacedSender", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + // Mirrors Transport.sendIntent, which reports whether the socket took it. + function record(sent: number[], id: number) { + return () => { + sent.push(id); + return true; + }; + } + + test("sends the first item immediately", () => { + const sender = new PacedSender(INTERVAL); + const sent: number[] = []; + + sender.push(record(sent, 0)); + + expect(sent).toEqual([0]); + }); + + test("holds the rest until their turn", () => { + const sender = new PacedSender(INTERVAL); + const sent: number[] = []; + + for (let i = 0; i < 4; i++) sender.push(record(sent, i)); + + expect(sent).toEqual([0]); + vi.advanceTimersByTime(INTERVAL); + expect(sent).toEqual([0, 1]); + vi.advanceTimersByTime(INTERVAL * 2); + expect(sent).toEqual([0, 1, 2, 3]); + }); + + test("delivers a whole fleet order in order", () => { + const sender = new PacedSender(INTERVAL); + const sent: number[] = []; + const batches = 15; + + for (let i = 0; i < batches; i++) sender.push(record(sent, i)); + vi.advanceTimersByTime(INTERVAL * batches); + + expect(sent).toEqual(Array.from({ length: batches }, (_, i) => i)); + }); + + test("never sends faster than the interval", () => { + const sender = new PacedSender(INTERVAL); + const times: number[] = []; + + for (let i = 0; i < 15; i++) { + sender.push(() => { + times.push(Date.now()); + return true; + }); + } + vi.advanceTimersByTime(INTERVAL * 15); + + for (let i = 1; i < times.length; i++) { + expect(times[i] - times[i - 1]).toBeGreaterThanOrEqual(INTERVAL); + } + }); + + test("keeps unsent batches queued while disconnected", () => { + const sender = new PacedSender(INTERVAL); + const sent: number[] = []; + let connected = false; + + for (let i = 0; i < 3; i++) { + sender.push(() => { + if (!connected) return false; + sent.push(i); + return true; + }); + } + + vi.advanceTimersByTime(INTERVAL * 10); + expect(sent).toEqual([]); + + connected = true; + vi.advanceTimersByTime(INTERVAL * 3); + expect(sent).toEqual([0, 1, 2]); + }); + + test("retries the failed batch before the ones behind it", () => { + const sender = new PacedSender(INTERVAL); + const order: number[] = []; + let failFirst = true; + + sender.push(() => { + if (failFirst) return false; + order.push(0); + return true; + }); + sender.push(record(order, 1)); + + expect(order).toEqual([]); + failFirst = false; + vi.advanceTimersByTime(INTERVAL * 2); + + expect(order).toEqual([0, 1]); + }); + + test("clear() drops anything still queued", () => { + const sender = new PacedSender(INTERVAL); + const sent: number[] = []; + + for (let i = 0; i < 10; i++) sender.push(record(sent, i)); + sender.clear(); + vi.advanceTimersByTime(INTERVAL * 20); + + expect(sent).toEqual([0]); + }); + + test("an interval of zero still preserves order", () => { + const sender = new PacedSender(0); + const sent: number[] = []; + + for (let i = 0; i < 5; i++) sender.push(record(sent, i)); + vi.runAllTimers(); + + expect(sent).toEqual([0, 1, 2, 3, 4]); + }); +});