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