Make the filter limit a restart setting with 0 for no limit

Lowering the limit at runtime would leave connections holding more
filters than allowed, so read it once when the controller starts.
Move the limit tests into the subscribe and replace groups.

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

docs/configuration.md

@@ -178,7 +178,8 @@ Defaults to 8.
 
 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.
+Set to 0 for no limit.
+Lowering the limit does not remove filters connections already hold.
 
 Defaults to 10000.
 

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(() => this.config.get("controller.max_subscription_filters"));
+	subscriptions: lib.SubscriptionController;
 	/** 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 */
@@ -273,6 +273,7 @@ export default class Controller {
 		});
 
 		// Handle subscriptions for all internal properties
+		this.subscriptions = new lib.SubscriptionController(config.get("controller.max_subscription_filters"));
 		this.subscriptions.handle(lib.SystemInfoUpdateEvent, this.handleSystemInfoSubscription.bind(this));
 		this.subscriptions.handle(lib.HostUpdatesEvent, this.handleHostSubscription.bind(this));
 		this.subscriptions.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsSubscription.bind(this));

packages/lib/src/config/definitions.ts

@@ -211,10 +211,11 @@ export class ControllerConfig extends classes.Config<ControllerConfigFields> {
 		},
 		"controller.max_subscription_filters": {
 			title: "Max Subscription Filters",
-			description: "Maximum number of filters a single connection can hold for one subscribed event.",
+			description: "Maximum number of filters a connection can hold for one subscribed event, 0 for no limit.",
 			type: "number",
 			initialValue: 10000,
-			validator: validators.all(validators.integer, validators.greaterThanZero),
+			restartRequired: true,
+			validator: validators.all(validators.integer, validators.greaterThanEqualZero),
 		},
 		"controller.system_metrics_interval": {
 			title: "System Metrics Interval",

packages/lib/src/subscriptions.ts

@@ -256,11 +256,11 @@ 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.
+	 * @param _maxFilters -
+	 *     Max number of filters a subscriber can hold for one event, 0 for no limit.
 	 */
 	constructor(
-		private _maxFilters: () => number = () => Infinity,
+		private _maxFilters = 0,
 	) { }
 
 	/**
@@ -338,7 +338,7 @@ export class SubscriptionController {
 	}
 
 	private _checkFilterCount(filters: SubscriptionFilters, existing?: SubscriptionFilters) {
-		if (existing?.isAll()) {
+		if (this._maxFilters === 0 || existing?.isAll()) {
 			return; // union into all is a noop
 		}
 		let count = filters.size;
@@ -350,8 +350,8 @@ export class SubscriptionController {
 				}
 			}
 		}
-		if (count > this._maxFilters()) {
-			throw new RequestError(`Subscription exceeds the limit of ${this._maxFilters()} filters`);
+		if (count > this._maxFilters) {
+			throw new RequestError(`Subscription exceeds the limit of ${this._maxFilters} filters`);
 		}
 	}
 

test/lib/subscriptions.test.js

@@ -730,50 +730,63 @@ describe("lib/subscriptions", function() {
 					await onceConnectorSend(0);
 					assertLastEvent(0, RegisteredEvent);
 				});
-			});
 
-			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() {
+				it("should reject filters over the limit", async function() {
 					const link = getLink(0);
-					await subscriptions.handleRequest(link, request("subscribe", ["a", "b"]), connectorData.dst);
+					const connectorData = connectorSetupDate[0];
+					const ownSubscriptions = new lib.SubscriptionController(2);
+					ownSubscriptions.handle(RegisteredEvent);
+
+					const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
+					await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
+
+					const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "baz");
 					await assert.rejects(
-						subscriptions.handleRequest(link, request("subscribe", ["c"]), connectorData.dst),
+						ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src),
 						lib.RequestError,
 					);
-					subscriptions.broadcast(new RegisteredEvent(), "c");
+
+					ownSubscriptions.broadcast(new RegisteredEvent(), "baz");
 					await new Promise(r => setImmediate(r));
 					assertNoEvent(0);
 				});
 
-				it("should not count filters already subscribed to", async function() {
+				it("should not count existing filters towards the limit", 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);
+					const connectorData = connectorSetupDate[0];
+					const ownSubscriptions = new lib.SubscriptionController(2);
+					ownSubscriptions.handle(RegisteredEvent);
+
+					const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
+					await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
+
+					const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
+					await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
 				});
 
-				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 not limit filters when subscribed to all", async function() {
+					const link = getLink(0);
+					const connectorData = connectorSetupDate[0];
+					const ownSubscriptions = new lib.SubscriptionController(2);
+					ownSubscriptions.handle(RegisteredEvent);
+
+					const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
+					await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
+
+					const request1 = new lib.SubscriptionRequest(
+						RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"]
 					);
+					await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
 				});
 
-				it("should allow subscribing to all", async function() {
+				it("should not limit filters when the limit is 0", 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);
+					const connectorData = connectorSetupDate[0];
+
+					const request = new lib.SubscriptionRequest(
+						RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"]
+					);
+					await subscriptions.handleRequest(link, request, connectorData.dst, connectorData.src);
 				});
 			});
 
@@ -897,6 +910,21 @@ describe("lib/subscriptions", function() {
 					await new Promise(r => setImmediate(r));
 					assertNoEvent(0);
 				});
+
+				it("should reject filters over the limit", async function() {
+					const link = getLink(0);
+					const connectorData = connectorSetupDate[0];
+					const ownSubscriptions = new lib.SubscriptionController(2);
+					ownSubscriptions.handle(RegisteredEvent);
+
+					const request = new lib.SubscriptionRequest(
+						RegisteredEvent.name, "replace", 0, ["foo", "bar", "baz"]
+					);
+					await assert.rejects(
+						ownSubscriptions.handleRequest(link, request, connectorData.dst, connectorData.src),
+						lib.RequestError,
+					);
+				});
 			});
 		});
 	});