From 690b79d2b744a74f79506aaa26599ba45b264915 Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:10:03 +0000 Subject: [PATCH 1/2] 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 --- docs/configuration.md | 9 +++++ packages/controller/src/Controller.ts | 2 +- packages/lib/src/config/definitions.ts | 8 ++++ packages/lib/src/subscriptions.ts | 28 ++++++++++++++ test/lib/subscriptions.test.js | 53 ++++++++++++++++++++++++-- 5 files changed, 95 insertions(+), 5 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 818f0e48..31c13c7c 100644 --- a/docs/configuration.md +++ b/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. diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 75a0e084..da0b12f0 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -83,7 +83,7 @@ export default class Controller { private _events = new events.EventEmitter(); /** 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 */ diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 27042c44..a8558b73 100644 --- a/packages/lib/src/config/definitions.ts +++ b/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 { 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", diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 298bf5d7..0f865827 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -255,6 +255,14 @@ type EventData = { export class SubscriptionController { _events = new Map(); + /** + * @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; diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 780d58b8..11ec9283 100644 --- a/test/lib/subscriptions.test.js +++ b/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"); From d2163cf6a0ba1eae2bd1328073b28a4096f857d4 Mon Sep 17 00:00:00 2001 From: bbassie <17990055+bbassie@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:13:36 +0000 Subject: [PATCH 2/2] 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 --- docs/configuration.md | 3 +- packages/controller/src/Controller.ts | 3 +- packages/lib/src/config/definitions.ts | 5 +- packages/lib/src/subscriptions.ts | 12 ++-- test/lib/subscriptions.test.js | 84 +++++++++++++++++--------- 5 files changed, 69 insertions(+), 38 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 31c13c7c..843b7831 100644 --- a/docs/configuration.md +++ b/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. diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index da0b12f0..a93301fd 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -83,7 +83,7 @@ export default class Controller { private _events = new events.EventEmitter(); /** 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)); diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index a8558b73..423f0a99 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -211,10 +211,11 @@ export class ControllerConfig extends classes.Config { }, "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", diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 0f865827..bcad8b2a 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -256,11 +256,11 @@ export class SubscriptionController { _events = new Map(); /** - * @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`); } } diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 11ec9283..e605a0fc 100644 --- a/test/lib/subscriptions.test.js +++ b/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, + ); + }); }); }); });