Merge branch 'master' into fix/stream-reject-closes-connection

test/controller/WsServer.js

@@ -113,6 +113,58 @@ describe("controller/src/WsServer", function() {
 			});
 		});
 	});
+	describe("connection metrics", function() {
+		const req = { socket: { remoteAddress: "10.0.0.1", remoteFamily: "IPv4" }, headers: {} };
+		/** @type {WsServer} */
+		let wsServer;
+		before(function() {
+			const config = new ControllerConfig("controller");
+			config.set("controller.auth_secret", Buffer.from("TestSecretDoNotUse").toString("base64"));
+			wsServer = new Controller(lib.logger, [], config).wsServer;
+		});
+
+		function counter(name) {
+			return lib.defaultRegistry.collectors.find(collector => collector.metric.name === name);
+		}
+		function value(name, type) {
+			const collector = counter(`clusterio_controller_websocket_${name}_total`);
+			return type ? collector.labels(type).get() : collector.get();
+		}
+
+		[["host", lib.Address.host], ["control", lib.Address.control]].forEach(([type, addressType]) => {
+			it(`should count ready, drop, resume and timeout for ${type} sessions`, function() {
+				const before = Object.fromEntries(
+					["ready", "drop", "resume", "session_timeout"].map(name => [name, value(name, type)])
+				);
+				const [connector, sessionToken] = wsServer.createSession(new lib.Address(addressType, 1));
+				connector.ready(new MockWebSocket(), new lib.Address(addressType, 1), sessionToken, undefined);
+				assert.equal(value("ready", type), before.ready + 1);
+
+				connector._socket.terminate();
+				assert.equal(value("drop", type), before.drop + 1);
+
+				wsServer.resumeSession(new lib.ResumeData(sessionToken, 0), new MockWebSocket(), req);
+				assert.equal(value("resume", type), before.resume + 1);
+
+				connector._socket.terminate();
+				connector._timedOut();
+				assert.equal(value("session_timeout", type), before.session_timeout + 1);
+				assert.equal(value("drop", type), before.drop + 2);
+				assert.equal(value("ready", type), before.ready + 1);
+			});
+		});
+		it("should count resumes of unknown sessions as failed", function() {
+			const before = value("failed_resume");
+			const socket = new MockWebSocket();
+			wsServer.resumeSession(new lib.ResumeData("invalid", 0), socket, req);
+			try {
+				assert.equal(value("failed_resume"), before + 1);
+				assert.equal(socket.sent[0].type, "invalidate");
+			} finally {
+				socket.emit("message", "not json"); // ends the handshake and its timeout
+			}
+		});
+	});
 	describe(".registerHost()", function() {
 		const secret = Buffer.from("TestSecretDoNotUse");
 		const req = { socket: { remoteAddress: "10.0.0.1", remoteFamily: "IPv4" }, headers: {} };

test/lib/logging_utils.js

@@ -45,6 +45,19 @@ describe("lib/logging_utils.js", function() {
 				logIndex
 			);
 		});
+		it("should drop entries for files no longer present", async function() {
+			let tempDir = path.join("temp", "test", "log_index_prune");
+			await fs.rm(tempDir, { recursive: true, force: true });
+			await fs.cp(logDir, tempDir, { recursive: true });
+			let logIndex = await lib.LogIndex.load(tempDir);
+			await logIndex.buildIndex();
+			await fs.unlink(path.join(tempDir, "test-2021-12-10.log"));
+			await logIndex.buildIndex();
+			assert.deepEqual(
+				new Set(logIndex.index.keys()),
+				new Set(["test-2021-12-11.log", "test-2021-12-12.log"])
+			);
+		});
 		it("should handle loading broken index", async function() {
 			let tempDir = path.join("temp", "test", "log");
 			await fs.writeFile(path.join(tempDir, "index.json"), "Broken JSON");
@@ -53,6 +66,43 @@ describe("lib/logging_utils.js", function() {
 		});
 	});
 
+	describe("deleteOldLogs", function() {
+		let tempDir = path.join("temp", "test", "log_retention");
+		let files = [
+			"cluster-2026-10-01.log",
+			"cluster-2026-10-12.log",
+			"cluster-2026-10-13.log",
+			"cluster-2026-11-12.log",
+			"index.json",
+			"notes-2026-10-01.txt",
+			".0123abcd-audit.json",
+		];
+		let now = Date.parse("2026-11-12T08:00:00Z");
+		beforeEach(async function() {
+			await fs.rm(tempDir, { recursive: true, force: true });
+			await fs.mkdir(tempDir, { recursive: true });
+			for (let file of files) {
+				await fs.writeFile(path.join(tempDir, file), "");
+			}
+		});
+
+		it("should delete only dated log files past the retention", async function() {
+			let deleted = await lib.deleteOldLogs(tempDir, 30, now);
+			assert.deepEqual(deleted.sort(), ["cluster-2026-10-01.log", "cluster-2026-10-12.log"]);
+			assert.deepEqual(
+				(await fs.readdir(tempDir)).sort(),
+				files.filter(file => !deleted.includes(file)).sort(),
+			);
+		});
+		it("should keep everything with a retention of 0", async function() {
+			assert.deepEqual(await lib.deleteOldLogs(tempDir, 0, now), []);
+			assert.deepEqual((await fs.readdir(tempDir)).sort(), [...files].sort());
+		});
+		it("should ignore a missing directory", async function() {
+			assert.deepEqual(await lib.deleteOldLogs(path.join(tempDir, "missing"), 30, now), []);
+		});
+	});
+
 	let allLines = [];
 	let logIndex = null;
 	before(async function() {

test/lib/subscriptions.test.js

@@ -1021,6 +1021,47 @@ describe("lib/subscriptions", function() {
 			it("should handle the provided event", function() {
 				assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
 			});
+
+			it("should share one connect and close listener on the connector", function() {
+				const connector = mockControl.connector;
+				const connectListeners = connector.listenerCount("connect");
+				const closeListeners = connector.listenerCount("close");
+				const subscribers = [
+					new lib.EventSubscriber(StringPermissionEvent, mockControl),
+					new lib.EventSubscriber(FunctionPermissionEvent, mockControl),
+				];
+				assert.equal(subscribers.length, 2);
+				assert.equal(connector.listenerCount("connect"), connectListeners);
+				assert.equal(connector.listenerCount("close"), closeListeners);
+			});
+
+			it("should pass connect and close to every subscriber on the connector", function() {
+				const subscribers = [
+					eventSubscriber,
+					new lib.EventSubscriber(StringPermissionEvent, mockControl),
+					new lib.EventSubscriber(FunctionPermissionEvent, mockControl),
+				];
+				const received = subscribers.map(() => []);
+				subscribers.forEach((subscriber, index) => {
+					subscriber.handleConnectionEvent = event => { received[index].push(event); };
+				});
+				mockControl.connector.emit("close");
+				mockControl.connector.emit("connect");
+				assert.deepEqual(received, [["close", "connect"], ["close", "connect"], ["close", "connect"]]);
+			});
+
+			it("should only pass events to subscribers on the same connector", function() {
+				const otherControl = new MockControl(new MockConnector(
+					addr({ controlId: 1 }),
+					addr("controller"),
+				));
+				const other = new lib.EventSubscriber(RegisteredEvent, otherControl);
+				const received = [];
+				eventSubscriber.handleConnectionEvent = event => { received.push(["first", event]); };
+				other.handleConnectionEvent = event => { received.push(["other", event]); };
+				otherControl.connector.emit("close");
+				assert.deepEqual(received, [["other", "close"]]);
+			});
 		});
 
 		describe("subscribe()", function() {