Merge pull request #571 from Hornwitser/link-bugfixes

Link bugfixes

CHANGELOG.md

@@ -4,7 +4,8 @@
 
 ### Features
 
-- Added support for `color-settings` Mod settings (#568)[https://github.com/clusterio/clusterio/issues/568].
+- Added support for `color-settings` Mod settings [#568](https://github.com/clusterio/clusterio/issues/568).
+- Added `clusterio_controller_pending_requests` and `clusterio_host_pending_requests` metrics.
 
 ### Changes
 
@@ -12,6 +13,8 @@
 - Fixed ctl not exporting types used for adding commands to ctl plugins.
 - Fixed controller only plugins throwing an error about missing a config field on startup out when added to hosts.
 - Fixed installer incorrectly claiming the options --factorio-dir and --no-download-headless conflicts with each other.
+- Fixed a memory leak on every request sent over the WebSocket.
+- Fixed assertion error on closing or invalidating a link with messages forwarded from a virtual link.
 
 
 ## Version 2.0.0-alpha.14

packages/controller/controller.ts

@@ -37,6 +37,24 @@ import { version } from "./package.json";
 // globals
 let controller: Controller;
 
+void new lib.Gauge(
+	"clusterio_controller_pending_requests",
+	"Count of pending link requests currently waiting in memory on the controller.",
+	{
+		callback: (gauge) => {
+			let count = 0;
+			if (controller) {
+				for (const host of controller.wsServer.hostConnections.values()) {
+					count += host.pendingRequestCount;
+				}
+				for (const control of controller.wsServer.controlConnections.values()) {
+					count += control.pendingRequestCount;
+				}
+			}
+			gauge.set(count);
+		},
+	}
+);
 
 void new lib.Gauge(
 	"clusterio_controller_host_mapping",

packages/host/host.ts

@@ -26,6 +26,27 @@ import { ConsoleTransport, levels, logger } from "@clusterio/lib";
 
 import Host from "./src/Host";
 
+let host: Host | undefined;
+
+void new lib.Gauge(
+	"clusterio_host_pending_requests",
+	"Count of pending link requests currently waiting in memory on the host.",
+	{
+		labels: ["host_id"],
+		callback: (gauge) => {
+			if (!host) {
+				return;
+			}
+
+			let count = host.pendingRequestCount;
+			for (const connection of host.instanceConnections.values()) {
+				count += connection.pendingRequestCount;
+				count += connection.instance.pendingRequestCount;
+			}
+			gauge.labels(String(host.config.get("host.id"))).set(count);
+		},
+	}
+);
 
 export class HostConnector extends lib.WebSocketClientConnector {
 	constructor(
@@ -205,7 +226,7 @@ async function startHost() {
 	}
 
 	let hostConnector = new HostConnector(hostConfig, tlsCa, pluginInfos);
-	let host = new Host(
+	host = new Host(
 		hostConnector,
 		args.config,
 		hostConfig,
@@ -227,7 +248,7 @@ async function startHost() {
 
 		secondSigint = true;
 		logger.info("Caught interrupt signal, shutting down");
-		host.shutdown();
+		host!.shutdown();
 	});
 	let secondSigterm = false;
 	process.on("SIGTERM", () => {
@@ -240,11 +261,11 @@ async function startHost() {
 
 		secondSigterm = true;
 		logger.info("Caught termination signal, shutting down");
-		host.shutdown();
+		host!.shutdown();
 	});
 	process.on("SIGHUP", () => {
 		logger.info("Terminal closed, shutting down");
-		host.shutdown();
+		host!.shutdown();
 	});
 
 	try {
@@ -255,7 +276,7 @@ async function startHost() {
 	}
 
 	hostConnector.once("connect", () => {
-		logger.add(new lib.LinkTransport({ link: host }));
+		logger.add(new lib.LinkTransport({ link: host! }));
 	});
 
 	await hostConnector.connect();

packages/lib/src/link/link.ts

@@ -141,17 +141,30 @@ export class Link {
 		});
 	}
 
+	/**
+	 * Count of requests currently waiting for a response on this link
+	 */
+	get pendingRequestCount() {
+		return this._pendingRequests.size + this._forwardedRequests.size;
+	}
+
 	_clearPendingRequests(err: Error & { code?: string }) {
 		for (let pending of this._pendingRequests.values()) {
 			pending.reject(err);
 		}
 		for (let pending of this._forwardedRequests.values()) {
-			assert(pending.origin.connector instanceof WebSocketBaseConnector);
-			if (pending.origin.connector.hasSession) {
-				pending.origin.connector.sendResponseError(
-					new libData.ResponseError(err.message, err.code), pending.src, pending.dst
-				);
+			// Drop response if the origin is a WebSocket and the session
+			// has expired.  The other end of the link will have sent an
+			// error response to the request in this case.
+			if (
+				pending.origin.connector instanceof WebSocketBaseConnector
+				&& !pending.origin.connector.hasSession
+			) {
+				continue;
 			}
+			pending.origin.connector.sendResponseError(
+				new libData.ResponseError(err.message, err.code), pending.src, pending.dst
+			);
 		}
 		this._pendingRequests.clear();
 		this._forwardedRequests.clear();
@@ -421,6 +434,7 @@ export class Link {
 			throw new libErrors.InvalidMessage(`Received reply from ${message.src} for message sent to ${pending.dst}`);
 		}
 
+		this._pendingRequests.delete(message.dst.requestId!);
 		try {
 			pending.resolve(pending.request.responseFromJSON(message.data));
 		} catch (err: any) {
@@ -437,6 +451,7 @@ export class Link {
 			);
 		}
 
+		this._pendingRequests.delete(message.dst.requestId!);
 		pending.reject(new libErrors.RequestError(message.data.message, message.data.code, message.data.stack));
 	}
 

test/lib/link/link.js

@@ -16,6 +16,7 @@ describe("lib/link/link", function() {
 
 	describe("class Link", function() {
 		let testConnector;
+		/** @type {lib.Link} */
 		let testLink;
 		let src = addr({ controlId: 1 });
 		let dst = addr("controller");
@@ -281,6 +282,7 @@ describe("lib/link/link", function() {
 				let srcReq = new lib.Address(lib.Address.control, 1, 1);
 				testConnector.emit("message", new lib.MessageResponse(1, dst, srcReq));
 				await request;
+				assert.deepEqual(testLink._pendingRequests, new Map());
 			});
 			it("should send request and reject with error when error response is received", async function() {
 				let message = events.once(testConnector, "send");
@@ -297,6 +299,7 @@ describe("lib/link/link", function() {
 					request,
 					{ message: "Error" }
 				);
+				assert.deepEqual(testLink._pendingRequests, new Map());
 			});
 			it("should send request and resolve with data when response is received", async function() {
 				let message = events.once(testConnector, "send");
@@ -305,6 +308,7 @@ describe("lib/link/link", function() {
 				let srcReq = new lib.Address(lib.Address.control, 1, 1);
 				testConnector.emit("message", new lib.MessageResponse(1, dst, srcReq, 44));
 				assert.deepEqual(await request, new NumberRequest.Response(44));
+				assert.deepEqual(testLink._pendingRequests, new Map());
 			});
 			it("should send event to the other side of the link", async function() {
 				let message = events.once(testConnector, "send");