Batch move_warship intents to stay under MAX_INTENT_SIZE

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",

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<typeof setTimeout> | 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);
+  }
+}

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,7 @@ export class Transport {
   private onmessage: (msg: ServerMessage) => void;
 
   private pingInterval: number | null = null;
+  private readonly warshipBatches: PacedSender;
   public readonly isLocal: boolean;
 
   constructor(
@@ -213,6 +221,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),
@@ -441,6 +453,7 @@ export class Transport {
   }
 
   leaveGame() {
+    this.warshipBatches.clear();
     if (this.isLocal) {
       this.localServer.endGame();
       return;
@@ -649,11 +662,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 +698,21 @@ export class Transport {
     this.sendIntent({ type: "toggle_game_start_timer" });
   }
 
-  private sendIntent(intent: Intent) {
+  private sendIntent(intent: Intent): boolean {
     if (this.isLocal || 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) {

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(),

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";
 

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);
+    }
+  });
+});

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]);
+  });
+});