fix(tracker): limit segment reports per peer across REST and WebSocket

API_REFERENCE.md

@@ -355,7 +355,14 @@ before additions, as for `segment_inventory_delta`. Updates from a peer that
 left before the batch was sent are dropped, so `peer_left` is never followed by
 a stale update. A batch carries at most 256 entries; larger batches are split.
 
-Malformed frames are rejected. Connections are closed during tracker shutdown;
+Malformed frames are rejected. Frames are limited to 64 KiB and each
+connection to 20 messages per second (bursts of 40). A segment report carries
+at most 512 IDs of up to 512 characters, and each peer may report 16 K
+characters of segment IDs per second (bursts of 256 K, enough to re-report a
+full inventory after reconnecting), counted across WebSocket and REST
+reports. A connection over a limit is closed with code `1008`; a REST report
+over the rate gets `429`. The tracker keeps at most 2000 segments and 128 KiB
+of segment IDs per peer, dropping the oldest. Connections are closed during tracker shutdown;
 clients should reconnect with exponential backoff and resubscribe.
 
 ## Browser SDK events

CHANGELOG.md

@@ -113,6 +113,12 @@ All notable changes to OpenStreamGrid are documented in this file.
 
 ### Changed
 
+- Each peer may report at most 16 K characters of segment IDs per second
+  (bursts of 256 K), across WebSocket and REST; beyond that its WebSocket is
+  closed with `1008` and REST reports get `429`. REST segment reports also get
+  the WebSocket limits of 512 IDs of up to 512 characters, where they were
+  unbounded. Reports are relayed to every neighbor, so these limits bound what
+  one peer can make the tracker send.
 - The tracker keeps at most 128 KiB of segment-ID characters per peer, in
   addition to 2000 segments, dropping the oldest. Up to 2000 IDs of 512
   characters let one peer carry about a megabyte of inventory into every peer

tracker/src/inventory-limits.ts

@@ -0,0 +1,85 @@
+/**
+ * Limits on segment inventory reports. The tracker relays every report to the
+ * reporter's neighbors (or to every subscriber, for legacy clients), so both
+ * the size of one report and the volume a peer reports over time are bounded,
+ * whichever interface (REST or WebSocket) the report arrives on.
+ */
+
+export const MAX_SEGMENTS_PER_REPORT = 512;
+export const MAX_SEGMENT_ID_LENGTH = 512;
+/**
+ * Sustained segment-ID characters a peer may report per second, with bursts
+ * large enough to re-report a full inventory after reconnecting. Clients
+ * report a few IDs per segment, far below this.
+ */
+export const INVENTORY_CHARACTERS_PER_SECOND = 16 * 1024;
+export const INVENTORY_BURST_CHARACTERS = 256 * 1024;
+/** Buckets idle this long are full again and can be forgotten. */
+const IDLE_BUCKET_MS = Math.ceil(
+  (INVENTORY_BURST_CHARACTERS / INVENTORY_CHARACTERS_PER_SECOND) * 1_000,
+);
+const SWEEP_EVERY_CONSUMES = 1_024;
+
+/** The segment IDs in `value`, or undefined unless it is a valid report list. */
+export const segmentIdList = (value: unknown): string[] | undefined => {
+  if (
+    !Array.isArray(value) ||
+    value.length > MAX_SEGMENTS_PER_REPORT ||
+    value.some(
+      (segment) =>
+        typeof segment !== "string" ||
+        segment.length === 0 ||
+        segment.length > MAX_SEGMENT_ID_LENGTH,
+    )
+  ) {
+    return undefined;
+  }
+  return value as string[];
+};
+
+interface InventoryBucket {
+  characters: number;
+  updatedAt: number;
+}
+
+/** Token bucket of reported segment-ID characters per broadcast peer. */
+export class InventoryRateLimiter {
+  private readonly buckets = new Map<string, InventoryBucket>();
+  private consumesSinceSweep = 0;
+
+  constructor(
+    private readonly charactersPerSecond = INVENTORY_CHARACTERS_PER_SECOND,
+    private readonly burstCharacters = INVENTORY_BURST_CHARACTERS,
+    private readonly now: () => number = Date.now,
+  ) {}
+
+  /** Charges the IDs in `segmentLists`; false (charging nothing) when over the limit. */
+  consume(broadcastId: string, peerId: string, segmentLists: readonly (readonly string[])[]): boolean {
+    let characters = 0;
+    for (const segments of segmentLists) {
+      for (const segment of segments) characters += segment.length;
+    }
+    const now = this.now();
+    this.sweep(now);
+    const key = `${broadcastId}\n${peerId}`;
+    const bucket = this.buckets.get(key) ?? { characters: this.burstCharacters, updatedAt: now };
+    bucket.characters = Math.min(
+      this.burstCharacters,
+      bucket.characters + ((now - bucket.updatedAt) / 1_000) * this.charactersPerSecond,
+    );
+    bucket.updatedAt = now;
+    this.buckets.set(key, bucket);
+    if (characters > bucket.characters) return false;
+    bucket.characters -= characters;
+    return true;
+  }
+
+  private sweep(now: number): void {
+    this.consumesSinceSweep += 1;
+    if (this.consumesSinceSweep < SWEEP_EVERY_CONSUMES) return;
+    this.consumesSinceSweep = 0;
+    for (const [key, bucket] of this.buckets) {
+      if (now - bucket.updatedAt >= IDLE_BUCKET_MS) this.buckets.delete(key);
+    }
+  }
+}

tracker/src/server.ts

@@ -23,6 +23,12 @@ import {
   type PeerSessionClaims,
 } from "./peer-session.js";
 import { clientAddress } from "./client-address.js";
+import {
+  InventoryRateLimiter,
+  MAX_SEGMENT_ID_LENGTH,
+  MAX_SEGMENTS_PER_REPORT,
+  segmentIdList,
+} from "./inventory-limits.js";
 import {
   issueTurnCredentials,
   TURN_CREDENTIAL_GRACE_MS,
@@ -936,14 +942,6 @@ const optionalMetadata = (
   return metadata;
 };
 
-const stringArray = (body: JsonObject, key: string): string[] => {
-  const value = body[key];
-  if (!Array.isArray(value) || value.some((item) => typeof item !== "string")) {
-    throw new RequestError(`'${key}' must be an array of strings`);
-  }
-  return value;
-};
-
 const decodedPathComponent = (value: string): string => {
   try {
     return decodeURIComponent(value);
@@ -977,6 +975,8 @@ export interface TrackerHttpOptions {
   maxPeersPerBroadcast?: number;
   metrics?: TrackerMetrics;
   rateLimiter?: IpRateLimiter;
+  /** Shared with the WebSocket hub so a peer's reports count once across both. */
+  inventoryLimiter?: InventoryRateLimiter;
   turn?: TurnCredentialOptions;
   /** Reverse proxies in front of the tracker whose X-Forwarded-For is trusted. */
   trustProxyHops?: number;
@@ -995,6 +995,7 @@ export const createTrackerHandler = (
   options: TrackerHttpOptions = {},
 ) => {
   const metrics = options.metrics ?? new TrackerMetrics();
+  const inventoryLimiter = options.inventoryLimiter ?? new InventoryRateLimiter();
   const rateLimiter =
     options.rateLimiter ??
     new IpRateLimiter(
@@ -1169,8 +1170,17 @@ export const createTrackerHandler = (
         if (action === "segments" && method === "POST") {
           requirePeerSession(broadcastId, peerId);
           const replace = optionalBoolean(body, "replace");
+          const segments = segmentIdList(body.segments);
+          if (!segments) {
+            throw new RequestError(
+              `'segments' must be an array of at most ${MAX_SEGMENTS_PER_REPORT} non-empty strings of up to ${MAX_SEGMENT_ID_LENGTH} characters`,
+            );
+          }
+          if (!inventoryLimiter.consume(broadcastId, peerId, [segments])) {
+            throw new RequestError("Segment report rate limit exceeded", 429);
+          }
           const report: SegmentPossessionReport = {
-            segments: stringArray(body, "segments"),
+            segments,
             ...(replace !== undefined ? { replace } : {}),
           };
           const peer = store.reportSegments(
@@ -1473,8 +1483,10 @@ export class TrackerServer {
       broadcastListChanged: () => webSockets?.broadcastListChanged(),
       broadcastRemoved: (broadcastId) => webSockets?.broadcastRemoved(broadcastId),
     };
+    const inventoryLimiter = new InventoryRateLimiter();
     this.server = createServer(
       createTrackerHandler(store, events, this.statsEvents, {
+        inventoryLimiter,
         rateLimitRps: configuration.rateLimitRps,
         rateLimitBurst: configuration.rateLimitBurst,
         maxPeersPerBroadcast: configuration.maxPeersPerBroadcast,
@@ -1500,6 +1512,7 @@ export class TrackerServer {
       this.statsEvents,
       peerSessions,
       {
+        inventoryLimiter,
         ...(configuration.wsAllowedOrigins
           ? { allowedOrigins: configuration.wsAllowedOrigins }
           : {}),

tracker/src/websocket.ts

@@ -18,6 +18,7 @@ import {
   bearerToken,
   type PeerSessionClaims,
 } from "./peer-session.js";
+import { InventoryRateLimiter, segmentIdList } from "./inventory-limits.js";
 import { NeighborViews, type NeighborViewFill } from "./neighbor-views.js";
 import type { TrackerStoreBackend } from "./store.js";
 
@@ -74,8 +75,6 @@ const VIEWERS_PER_FLUSH_TURN = 256;
 const MAX_SUBSCRIBE_FEATURES = 16;
 const MAX_BUFFERED_WEB_SOCKET_BYTES = 1024 * 1024;
 const MAX_WEB_SOCKET_PAYLOAD_BYTES = 64 * 1024;
-const MAX_SEGMENTS_PER_REPORT = 512;
-const MAX_SEGMENT_ID_LENGTH = 512;
 const MESSAGE_RATE_PER_SECOND = 20;
 const MESSAGE_BURST = 40;
 const MAX_CONNECTIONS_PER_IP = 20;
@@ -144,19 +143,8 @@ const optionalUnitInterval = (
 };
 
 const requiredSegments = (message: JsonObject): string[] => {
-  const segments = message.segments;
-  if (
-    !Array.isArray(segments) ||
-    segments.length > MAX_SEGMENTS_PER_REPORT ||
-    segments.some(
-      (segment) =>
-        typeof segment !== "string" ||
-        segment.length === 0 ||
-        segment.length > MAX_SEGMENT_ID_LENGTH,
-    )
-  ) {
-    throw new Error("'segments' must be an array of strings");
-  }
+  const segments = segmentIdList(message.segments);
+  if (!segments) throw new Error("'segments' must be an array of strings");
   return segments;
 };
 
@@ -185,6 +173,8 @@ export interface TrackerWebSocketOptions {
   neighborViewSize?: number;
   /** Neighbor-view subscribers served per event-loop turn (default 256). */
   viewersPerFlushTurn?: number;
+  /** Shared with the REST API so a peer's reports count once across both. */
+  inventoryLimiter?: InventoryRateLimiter;
 }
 
 /** WebSocket fan-out counters exported as Prometheus metrics. */
@@ -213,6 +203,7 @@ export class TrackerWebSocketHub implements TrackerEvents {
   private readonly flushRequested = new Set<string>();
   private readonly sessions = new Map<WebSocket, PeerSessionClaims>();
   private readonly messageBudgets = new Map<WebSocket, MessageBudget>();
+  private readonly inventoryLimiter: InventoryRateLimiter;
   private readonly connectionIps = new Map<WebSocket, string>();
   private readonly connectionsPerIp = new Map<string, number>();
   private readonly sessionExpiryTimers = new Map<WebSocket, NodeJS.Timeout>();
@@ -232,6 +223,7 @@ export class TrackerWebSocketHub implements TrackerEvents {
     private readonly peerSessions = new PeerSessionTokenService(),
     private readonly options: TrackerWebSocketOptions = {},
   ) {
+    this.inventoryLimiter = options.inventoryLimiter ?? new InventoryRateLimiter();
     const neighborViewSize = options.neighborViewSize ?? DEFAULT_NEIGHBOR_VIEW_SIZE;
     this.neighborViews =
       neighborViewSize > 0 ? new NeighborViews(neighborViewSize) : undefined;
@@ -552,6 +544,9 @@ export class TrackerWebSocketHub implements TrackerEvents {
         if (parsed.added !== undefined || parsed.removed !== undefined) {
           const added = requiredSegments({ segments: parsed.added ?? [] });
           const removed = requiredSegments({ segments: parsed.removed ?? [] });
+          if (!this.inventoryLimiter.consume(broadcastId, peerId, [added, removed])) {
+            throw new Error("Segment report rate limit exceeded");
+          }
           this.store.applySegmentDelta(broadcastId, peerId, added, removed);
           this.publishInventory({
             type: "segment_inventory_delta",
@@ -566,6 +561,9 @@ export class TrackerWebSocketHub implements TrackerEvents {
         if (parsed.replace !== undefined && typeof parsed.replace !== "boolean") {
           throw new Error("'replace' must be a boolean");
         }
+        if (!this.inventoryLimiter.consume(broadcastId, peerId, [segments])) {
+          throw new Error("Segment report rate limit exceeded");
+        }
         const replace = parsed.replace;
         if (replace && this.hasSameSegments(broadcastId, peerId, segments)) return;
         this.store.reportSegments(broadcastId, peerId, segments, replace);

tracker/test/inventory-limits.test.ts

@@ -0,0 +1,39 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import {
+  InventoryRateLimiter,
+  MAX_SEGMENT_ID_LENGTH,
+  MAX_SEGMENTS_PER_REPORT,
+  segmentIdList,
+} from "../src/inventory-limits.js";
+
+test("accepts only bounded lists of non-empty segment IDs", () => {
+  assert.deepEqual(segmentIdList(["a.ts", "b.ts"]), ["a.ts", "b.ts"]);
+  assert.deepEqual(segmentIdList([]), []);
+  assert.equal(segmentIdList("a.ts"), undefined);
+  assert.equal(segmentIdList([1]), undefined);
+  assert.equal(segmentIdList([""]), undefined);
+  assert.equal(segmentIdList(["x".repeat(MAX_SEGMENT_ID_LENGTH + 1)]), undefined);
+  assert.equal(segmentIdList(Array.from({ length: MAX_SEGMENTS_PER_REPORT + 1 }, () => "a.ts")), undefined);
+});
+
+test("limits reported characters per peer with a refilling burst", () => {
+  let now = 0;
+  const limiter = new InventoryRateLimiter(100, 1_000, () => now);
+  const ids = (characters: number): string[] => ["x".repeat(characters)];
+
+  assert.equal(limiter.consume("live", "a", [ids(600)]), true);
+  assert.equal(limiter.consume("live", "a", [ids(300), ids(100)]), true);
+  // The burst is spent, and an over-budget report charges nothing.
+  assert.equal(limiter.consume("live", "a", [ids(1)]), false);
+  // Other peers and broadcasts have their own budgets.
+  assert.equal(limiter.consume("live", "b", [ids(1_000)]), true);
+  assert.equal(limiter.consume("other", "a", [ids(1_000)]), true);
+
+  now = 2_000; // refills 200 characters
+  assert.equal(limiter.consume("live", "a", [ids(250)]), false);
+  assert.equal(limiter.consume("live", "a", [ids(200)]), true);
+  now = 60_000; // never beyond the burst
+  assert.equal(limiter.consume("live", "a", [ids(1_001)]), false);
+  assert.equal(limiter.consume("live", "a", [ids(1_000)]), true);
+});