Share one connect and close listener between subscribers (#1171)

* Share one connect and close listener between subscribers

Every EventSubscriber added its own connect and close listener to the
connector, and Link caps the connector at 20 listeners per event. The
web Control and App use 13 close listeners, so the 8th plugin
subscriber made the browser log MaxListenersExceededWarning on every
page load. Subscribers on a connector now share one listener for each
event, which leaves the cap in place for everything else.

* Keep the connector subscribers as a static on EventSubscriber

packages/controller/src/WsServer.ts

@@ -30,6 +30,42 @@ const wsRejectedConnectionsCounter = new lib.Counter(
 	"How many WebSocket connections have been rejected during the handshake on the controller"
 );
 
+const wsReadyCounter = new lib.Counter(
+	"clusterio_controller_websocket_ready_total",
+	"How many WebSocket sessions have been established on the controller",
+	{ labels: ["type"] }
+);
+
+const wsResumeCounter = new lib.Counter(
+	"clusterio_controller_websocket_resume_total",
+	"How many WebSocket sessions have been resumed on the controller",
+	{ labels: ["type"] }
+);
+
+const wsDropCounter = new lib.Counter(
+	"clusterio_controller_websocket_drop_total",
+	"How many WebSocket sessions have lost their connection without closing on the controller",
+	{ labels: ["type"] }
+);
+
+const wsSessionTimeoutCounter = new lib.Counter(
+	"clusterio_controller_websocket_session_timeout_total",
+	"How many WebSocket sessions have timed out waiting for a resume on the controller",
+	{ labels: ["type"] }
+);
+
+const wsFailedResumeCounter = new lib.Counter(
+	"clusterio_controller_websocket_failed_resume_total",
+	"How many WebSocket resume attempts have been rejected due to an unknown or expired session on the controller"
+);
+
+for (const type of ["host", "control"]) {
+	wsReadyCounter.labels(type);
+	wsResumeCounter.labels(type);
+	wsDropCounter.labels(type);
+	wsSessionTimeoutCounter.labels(type);
+}
+
 /**
  * WebSocket server
  * @alias module:controller/src/WsServer
@@ -240,6 +276,12 @@ ${err.stack}`
 			this.activeConnectors.delete(sessionId);
 		});
 
+		const type = dst.type === lib.Address.host ? "host" : "control";
+		connector.on("connect", () => wsReadyCounter.labels(type).inc());
+		connector.on("resume", () => wsResumeCounter.labels(type).inc());
+		connector.on("drop", () => wsDropCounter.labels(type).inc());
+		connector.on("timeout", () => wsSessionTimeoutCounter.labels(type).inc());
+
 		return [connector, sessionToken];
 	}
 
@@ -367,6 +409,7 @@ ${err.stack}`
 			}
 
 		} catch (err) {
+			wsFailedResumeCounter.inc();
 			socket.send(JSON.stringify(new lib.MessageInvalidate()));
 			this.attachHandler(socket, req);
 			return;

packages/controller/src/WsServerConnector.ts

@@ -7,13 +7,17 @@ import * as lib from "@clusterio/lib";
 import { logger } from "@clusterio/lib";
 
 
+type WsServerConnectorEvents = {
+	"timeout": [],
+};
+
 /**
  * Connector for controller connections
  *
  * @extends module:lib.WebSocketBaseConnector
  * @alias module:controller/src/WsServerConnector
  */
-export default class WsServerConnector extends lib.WebSocketBaseConnector {
+export default class WsServerConnector extends lib.WebSocketBaseConnector<WsServerConnectorEvents> {
 	_timeoutId: ReturnType<typeof setTimeout> | null = null;
 
 	constructor(
@@ -119,6 +123,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector {
 
 	_timedOut() {
 		logger.verbose(`Connector | Connection to ${this.dst} timed out`);
+		this.emit("timeout");
 		this._close();
 	}
 

packages/create/create.js

@@ -46,6 +46,10 @@ const factorioLocations = {
 	],
 	darwin: [
 		"/Applications/factorio.app/Contents",
+		path.join(
+			os.homedir(), "Library", "Application Support", "Steam", "steamapps", "common", "Factorio",
+			"factorio.app", "Contents",
+		),
 	],
 };
 

packages/lib/src/subscriptions.ts

@@ -460,6 +460,39 @@ export interface SubscribableValue {
  * Multiple handlers can be subscribed at the same time
  */
 export class EventSubscriber<E, S = null> {
+	/** Subscribers created on each connector, they share one connect and one close listener on it */
+	private static connectorSubscribers = new WeakMap<object, Set<EventSubscriber<any, any>>>();
+
+	/**
+	 * Pass connect and close events of a connector on to a subscriber
+	 * Uses one listener for all subscribers, so that plugins creating many subscribers do not go past the
+	 * listener limit set on the connector.
+	 * @param connector - connector of the link the subscriber was created on
+	 * @param subscriber - subscriber to pass the events to
+	 */
+	private static addConnectorSubscriber(
+		connector: WebSocketBaseConnector,
+		subscriber: EventSubscriber<any, any>,
+	) {
+		let subscribers = EventSubscriber.connectorSubscribers.get(connector);
+		if (!subscribers) {
+			const created = new Set<EventSubscriber<any, any>>();
+			EventSubscriber.connectorSubscribers.set(connector, created);
+			connector.on("connect", () => {
+				for (const entry of created) {
+					entry.handleConnectionEvent("connect");
+				}
+			});
+			connector.on("close", () => {
+				for (const entry of created) {
+					entry.handleConnectionEvent("close");
+				}
+			});
+			subscribers = created;
+		}
+		subscribers.add(subscriber);
+	}
+
 	/** The time at which an event was last received, is less than 0 when there have been no events */
 	lastUpdatedMs = -1;
 	/** True if this subscriber is currently synced with the source */
@@ -485,13 +518,7 @@ export class EventSubscriber<E, S = null> {
 	) {
 		control.handle(Event, this._handleEvent.bind(this));
 		// The events below exist only on websocket connectors, we can't early return because of test mocking
-		const webSocketConnector = control.connector as WebSocketBaseConnector;
+		EventSubscriber.addConnectorSubscriber(control.connector as WebSocketBaseConnector, this);
-		webSocketConnector.on("connect", () => {
-			this.handleConnectionEvent("connect");
-		});
-		webSocketConnector.on("close", () => {
-			this.handleConnectionEvent("close");
-		});
 		// Only the web Control reports account changes
 		const accountControl = control as Link & { onAccountUpdate?: (handler: () => void) => void };
 		if (typeof accountControl.onAccountUpdate === "function") {

packages/web_ui/src/components/InstanceList.tsx

@@ -7,6 +7,7 @@ import type { ColumnType } from "antd/es/table/interface";
 import { useAccount } from "../model/account";
 import { useSystems } from "../model/system";
 import { useHosts } from "../model/host";
+import { useUsers } from "../model/user";
 import { useDefaultModPack, useModPacks } from "../model/mod_pack";
 import { RestartRequired } from "./system_metrics";
 import InstanceStatusTag from "./InstanceStatusTag";
@@ -33,6 +34,17 @@ export default function InstanceList(props: InstanceListProps) {
 	const [systems] = useSystems();
 	const [modPacks] = useModPacks();
 	const defaultModPack = useDefaultModPack();
+	const [users] = useUsers();
+	const canUsers = account.hasPermission("core.user.subscribe");
+	const onlineCounts = React.useMemo(() => {
+		const counts = new Map<number, number>();
+		for (const user of users.values()) {
+			for (const instanceId of user.instances ?? []) {
+				counts.set(instanceId, (counts.get(instanceId) ?? 0) + 1);
+			}
+		}
+		return counts;
+	}, [users]);
 	const tableState = useTableQueryState<lib.InstanceDetails>({
 		namespace: "instance", defaultSortKey: "name",
 	});
@@ -146,6 +158,18 @@ export default function InstanceList(props: InstanceListProps) {
 		},
 	];
 
+	if (canUsers) {
+		const playerCount = (instance: lib.InstanceDetails) => onlineCounts.get(instance.id) ?? 0;
+		columns.push({
+			title: "Players",
+			key: "players",
+			render: (_, instance) => playerCount(instance),
+			sorter: (a, b) => playerCount(a) - playerCount(b),
+			sortOrder: tableState.sortOrder("players"),
+			responsive: ["md"],
+		});
+	}
+
 	if (account.hasAnyPermission(...InstanceControlButtonPermissions)) {
 		columns.push({
 			key: "action",

packages/web_ui/src/components/ModPackViewPage.tsx

@@ -182,6 +182,7 @@ function SearchModsTable(props: SearchModsTableProps) {
 
 	return <>
 		<Input
+			autoFocus
 			placeholder="Search Mods"
 			style={{ marginBottom: 8 }}
 			onChange={(event) => {

packages/web_ui/src/components/ModsPage.tsx

@@ -1,7 +1,8 @@
-import React, { Fragment, useState, useContext, useEffect, useMemo } from "react";
+import React, { Fragment, useState, useContext, useEffect, useMemo, useRef } from "react";
 import { useNavigate } from "react-router-dom";
 import {
-	Button, Form, Input, Modal, Popconfirm, Space, Table, Typography, Upload, Alert, Select, Tooltip, notification,
+	Button, Form, Input, InputRef, Modal, Popconfirm, Space, Table, Typography, Upload, Alert, Select, Tooltip,
+	notification,
 } from "antd";
 import {
 	ImportOutlined, PlusOutlined, SearchOutlined, DownloadOutlined,
@@ -147,6 +148,7 @@ function SearchModsButton() {
 	const account = useAccount();
 	const [open, setOpen] = useState(false);
 	const [form] = Form.useForm();
+	const nameInputRef = useRef<InputRef>(null);
 	const [searchText, setSearchText] = useState("");
 	const [factorioVersion, setFactorioVersion] = useState<lib.MajorMinorVersion>("2.0");
 
@@ -333,6 +335,7 @@ function SearchModsButton() {
 			title="Search Mod Portal"
 			open={open}
 			onCancel={() => { setOpen(false); }}
+			afterOpenChange={visible => { if (visible) { nameInputRef.current?.focus(); } }}
 			width={1000}
 			footer={[
 				<Button key="close" onClick={() => { setOpen(false); }}>
@@ -349,7 +352,7 @@ function SearchModsButton() {
 				initialValues={{ factorioVersion: factorioVersion, name: searchText }}
 			>
 				<Form.Item name="name" label="Filter by Name or Title">
-					<Input placeholder="Start typing to filter..." />
+					<Input ref={nameInputRef} placeholder="Start typing to filter..." />
 				</Form.Item>
 				<Form.Item
 					name="factorioVersion"

packages/web_ui/src/components/OnlinePlayersList.tsx

@@ -1,4 +1,5 @@
-import React from "react";
+import React, { useState } from "react";
+import { Collapse } from "antd";
 
 import { useUsers } from "../model/user";
 import UsersTable from "./UsersTable";
@@ -9,23 +10,26 @@ type OnlinePlayersListProps = {
 
 export default function OnlinePlayersList({ instanceId }: OnlinePlayersListProps) {
 	const [users] = useUsers();
+	// undefined until the user toggles the panel, then their choice sticks.
+	const [open, setOpen] = useState<boolean>();
 
-	// Filter users to only show those online on this specific instance
+	const onlineCount = [...users.values()].filter(user => user.instances?.has(instanceId)).length;
-	const onlineUsers = [...users.values()].filter(user => user.instances && user.instances.has(instanceId));
+	const expanded = open ?? onlineCount > 0;
 
-	if (onlineUsers.length === 0) {
+	return <Collapse
-		return null;
+		size="small"
-	}
+		style={{ marginTop: 16 }}
-
+		activeKey={expanded ? ["players"] : []}
-	return (
+		onChange={keys => setOpen(keys.length > 0)}
-		<div style={{ marginTop: 16 }}>
+		items={[{
-			<h4>Online Players ({onlineUsers.length})</h4>
+			key: "players",
-			<UsersTable instanceId={instanceId} onlyOnline size="small" pagination={{
+			label: `Online Players (${onlineCount})`,
+			children: <UsersTable instanceId={instanceId} onlyOnline size="small" pagination={{
 				defaultPageSize: 10,
 				showSizeChanger: true,
 				pageSizeOptions: ["10", "20", "50", "100", "200"],
 				showTotal: (total: number) => `${total} Users`,
-			}} />
+			}} />,
-		</div>
+		}]}
-	);
+	/>;
 }

plugins/player_sync/module/serialize.lua

@@ -1271,23 +1271,8 @@ local function ensure_character(player)
 		player.teleport(force.get_spawn_position(planet_surface), planet_surface)
 	end
 
-	-- create_character needs a free spot within 200 tiles and does not generate chunks to find one. It fails in a chunk
+	-- Create and return the character
-	-- which does not exist yet, such as one the player was teleported into a moment ago, and where water or void is all
-	-- there is. The character is then made at the force's spawn. From a missing chunk it is moved back, as the game
-	-- generates the chunks around a character. Otherwise there is nowhere to stand, and it stays at the spawn.
 	if not player.create_character() then
-		surface = player.surface
-		local position = player.position
-		local force = player.force --[[@as LuaForce]]
-		local missing_chunk = not surface.is_chunk_generated({ math.floor(position.x / 32), math.floor(position.y / 32) })
-		player.teleport(force.get_spawn_position(surface), surface)
-		if player.create_character() and missing_chunk then
-			player.teleport(position, surface)
-		end
-	end
-
-	-- Return the character
-	if not player.character then
 		error(string.format(
 			"Failed to create character (controller: %s, physical: %s, surface: %s, connected: %s, driving: %s)",
 			controller_to_name[player.controller_type],

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/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() {