Limit the number of subscription filters per subscriber

A subscriber could keep sending subscribe requests with new filters and
the controller would add every one of them to the subscriber's filter
set without bound, growing memory until it ran out.

Add controller.max_subscription_filters (default 10000) and reject
subscribe and replace requests that would leave a subscriber holding
more filters than that for one event.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/configuration.md

@@ -174,6 +174,15 @@ Must be greater than 0.
 Defaults to 8.
 
 
+### controller.max_subscription_filters
+
+Maximum number of filters one connection can hold for a single subscribed event.
+Subscription requests that would go over this limit are rejected.
+Must be an integer greater than 0.
+
+Defaults to 10000.
+
+
 ### controller.system_metrics_interval
 
 Interval in seconds the controller gathers CPU, memory and disk usage from itself and every connected host for display in the web interface.

packages/controller/src/Controller.ts

@@ -83,7 +83,7 @@ export default class Controller {
 	private _events = new events.EventEmitter<ControllerEvents>();
 
 	/** Event subscription controller */
-	subscriptions = new lib.SubscriptionController();
+	subscriptions = new lib.SubscriptionController(() => this.config.get("controller.max_subscription_filters"));
 	/** User manager for the cluster, also contains access to the user records */
 	users: UserManager;
 	/** Instance manager for the cluster, also contains access to the user records */

packages/lib/src/config/definitions.ts

@@ -26,6 +26,7 @@ export interface ControllerConfigFields {
 	"controller.heartbeat_interval": number;
 	"controller.session_timeout": number;
 	"controller.metrics_timeout": number;
+	"controller.max_subscription_filters": number;
 	"controller.system_metrics_interval": number;
 	"controller.proxy_stream_timeout": number;
 	"controller.factorio_username": string | null;
@@ -208,6 +209,13 @@ export class ControllerConfig extends classes.Config<ControllerConfigFields> {
 			initialValue: 8,
 			validator: validators.greaterThanZero,
 		},
+		"controller.max_subscription_filters": {
+			title: "Max Subscription Filters",
+			description: "Maximum number of filters a single connection can hold for one subscribed event.",
+			type: "number",
+			initialValue: 10000,
+			validator: validators.all(validators.integer, validators.greaterThanZero),
+		},
 		"controller.system_metrics_interval": {
 			title: "System Metrics Interval",
 			description: "Interval in seconds to collect and update system metrics for the Web UI",

packages/lib/src/subscriptions.ts

@@ -255,6 +255,14 @@ type EventData = {
 export class SubscriptionController {
 	_events = new Map<string, EventData>();
 
+	/**
+	 * @param maxFilters -
+	 *     Optional function returning the max number of filters a subscriber can hold for one event.
+	 */
+	constructor(
+		private _maxFilters: () => number = () => Infinity,
+	) { }
+
 	/**
 	 * Allow clients to subscribe to an event by telling the subscription controller to accept them
 	 * Has an optional subscription update handler which is called when a client subscribes
@@ -329,6 +337,24 @@ export class SubscriptionController {
 		}
 	}
 
+	private _checkFilterCount(filters: SubscriptionFilters, existing?: SubscriptionFilters) {
+		if (existing?.isAll()) {
+			return; // union into all is a noop
+		}
+		let count = filters.size;
+		if (existing) {
+			count = existing.size;
+			for (const filter of filters) {
+				if (!existing.has(filter)) {
+					count += 1;
+				}
+			}
+		}
+		if (count > this._maxFilters()) {
+			throw new RequestError(`Subscription exceeds the limit of ${this._maxFilters()} filters`);
+		}
+	}
+
 	/**
 	 * Handle incoming subscription requests on a link
 	 * @param link - Link message was received on
@@ -359,6 +385,7 @@ export class SubscriptionController {
 				break;
 
 			case "subscribe":
+				this._checkFilterCount(request.filters, subscriber?.filters);
 				if (!subscriber) {
 					eventData.subscriptions.set(index, { link, address, filters: request.filters });
 				} else {
@@ -367,6 +394,7 @@ export class SubscriptionController {
 				break;
 
 			case "replace":
+				this._checkFilterCount(request.filters);
 				if (request.filters.isEmpty()) {
 					if (!subscriber) {
 						return false;

test/lib/subscriptions.test.js

@@ -391,7 +391,7 @@ describe("lib/subscriptions", function() {
 					return [connectorData.id, new lib.Link(new MockConnector(connectorData.src, connectorData.dst))];
 				})),
 			};
-			subscriptions = new lib.SubscriptionController(mockController);
+			subscriptions = new lib.SubscriptionController();
 		});
 
 		describe(".handle()", function() {
@@ -628,7 +628,7 @@ describe("lib/subscriptions", function() {
 				}
 
 				it("should strip the request id from broadcast events", async function() {
-					const ownSubscriptions = new lib.SubscriptionController(mockController);
+					const ownSubscriptions = new lib.SubscriptionController();
 					ownSubscriptions.handle(RegisteredEvent);
 					await ownSubscriptions.handleRequest(
 						getLink(connectorData.id),
@@ -646,7 +646,7 @@ describe("lib/subscriptions", function() {
 				});
 
 				it("should strip the request id from the replayed event", async function() {
-					const ownSubscriptions = new lib.SubscriptionController(mockController);
+					const ownSubscriptions = new lib.SubscriptionController();
 					ownSubscriptions.handle(RegisteredEvent, async () => new RegisteredEvent());
 					await ownSubscriptions.handleRequest(
 						getLink(connectorData.id),
@@ -661,7 +661,7 @@ describe("lib/subscriptions", function() {
 				});
 
 				it("should strip the request id when replacing filters creates the subscription", async function() {
-					const ownSubscriptions = new lib.SubscriptionController(mockController);
+					const ownSubscriptions = new lib.SubscriptionController();
 					ownSubscriptions.handle(RegisteredEvent);
 					await ownSubscriptions.handleRequest(
 						getLink(connectorData.id),
@@ -732,6 +732,51 @@ describe("lib/subscriptions", function() {
 				});
 			});
 
+			describe("filter limit", function() {
+				const connectorData = connectorSetupDate[0];
+				function request(action, filters) {
+					return new lib.SubscriptionRequest(
+						RegisteredEvent.name, action, 0, lib.SubscriptionFilters.fromShorthand(filters)
+					);
+				}
+				beforeEach(function() {
+					subscriptions = new lib.SubscriptionController(() => 2);
+					subscriptions.handle(RegisteredEvent);
+				});
+
+				it("should reject a subscribe that goes over the limit", async function() {
+					const link = getLink(0);
+					await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst);
+					await assert.rejects(
+						subscriptions.handleRequest(link, request("subscribe", ["c"]), connectorData.dst),
+						lib.RequestError,
+					);
+					subscriptions.broadcast(new RegisteredEvent(), "c");
+					await new Promise(r => setImmediate(r));
+					assertNoEvent(0);
+				});
+
+				it("should not count filters already subscribed to", async function() {
+					const link = getLink(0);
+					await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst);
+					await subscriptions.handleRequest(link, request("subscribe", ["a"]), connectorData.dst);
+				});
+
+				it("should reject a replace that goes over the limit", async function() {
+					await assert.rejects(
+						subscriptions.handleRequest(getLink(0), request("replace", ["a", "b", "c"]), connectorData.dst),
+						lib.RequestError,
+					);
+				});
+
+				it("should allow subscribing to all", async function() {
+					const link = getLink(0);
+					await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst);
+					await subscriptions.handleRequest(link, request("subscribe"), connectorData.dst);
+					await subscriptions.handleRequest(link, request("subscribe", ["c", "d", "e"]), connectorData.dst);
+				});
+			});
+
 			describe("unsubscribe", function() {
 				it("should remove a subscription", async function() {
 					const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe");