Fix crash on continuing an active connection (#369)

Fix ServerConnector reaching an invalid state if a continue operation is
performed on an active connection.  The continue would cause the
connection established logic to be runned before the connection lost
logic of the pervious connection is perfomed, resulting in a leaked
heartbeat interval and the connection lost logic being performed on the
new connection instead of the old.  Prevent this by closing both the new
and the old connection in this case.

packages/lib/link/connectors.js

@@ -75,6 +75,9 @@ class WebSocketBaseConnector extends events.EventEmitter {
 	 * Start sending heartbeats
 	 */
 	startHeartbeat() {
+		if (this._heartbeatId) {
+			throw new Error("heartbeat is already running");
+		}
 		this._lastHeartbeat = Date.now();
 		this._heartbeatId = setInterval(() => {
 			this._doHeartbeat();

packages/master/master.js

@@ -1399,7 +1399,24 @@ class WebSocketServerConnector extends libLink.WebSocketBaseConnector {
 	 * @param {number} lastSeq - The last message the client received.
 	 */
 	continue(socket, lastSeq) {
-		this._socket.terminate();
+
+		// It's possible the previous connection hasn't closed yet due to a
+		// stale connection.  If this is the case the connector will have to
+		// wait for the previous socket to close down before it can continue
+		// from a new socket, or there will be overlapping events from both.
+		if (["connected", "closing"].includes(this._state)) {
+			this._socket.close(4003, "Session Hijacked");
+
+			// Correctly waiting for the connector to close and then
+			// handling all the edge cases like multiple continues happening
+			// in parallel or connections going stale is far too difficult.
+			// Kill the connection here and let the client reconnect later
+			// when the connector is not held up by an active connection.
+			socket.close(1013, "Session Busy");
+			socket.terminate();
+			return;
+		}
+
 		this._socket = socket;
 
 		if (this._timeoutId) {

test/integration/clusterio.js

@@ -2,11 +2,15 @@
 const assert = require("assert").strict;
 const fs = require("fs-extra");
 const path = require("path");
+const events = require("events");
 
 const libLink = require("@clusterio/lib/link");
 const libUsers = require("@clusterio/lib/users");
 
-const { slowTest, get, exec, execCtl, sendRcon, getControl, instancesDir } = require("./index");
+const {
+	TestControl, TestControlConnector, url, controlToken, slowTest,
+	get, exec, execCtl, sendRcon, getControl, instancesDir,
+} = require("./index");
 
 
 async function getInstances() {
@@ -32,6 +36,29 @@ describe("Integration of Clusterio", function() {
 				);
 			});
 		});
+
+		describe("run", function() {
+			it("should handle resume of an active connection", async function() {
+				slowTest(this);
+				let tlsCa = await fs.readFile("test/file/tls/cert.pem");
+				let connectorA = new TestControlConnector(url, 2, tlsCa);
+				connectorA.token = controlToken;
+				let controlA = new TestControl(connectorA);
+				await connectorA.connect();
+				connectorA._state = "closing";
+				connectorA.stopHeartbeat();
+				connectorA.on("error", () => {});
+
+				let connectorB = new TestControlConnector(url, 2, tlsCa);
+				connectorB.token = controlToken;
+				let controlB = new TestControl(connectorB);
+				connectorB._sessionToken = connectorA._sessionToken;
+				connectorB._state = "handshake";
+				await connectorB._doConnect();
+				await events.once(connectorB, "resume");
+				await connectorB.close();
+			});
+		});
 	});
 
 	describe("clusterioctl", function() {

test/integration/index.js

@@ -197,6 +197,7 @@ module.exports = {
 	getControl,
 
 	url,
+	controlToken,
 	instancesDir,
 	databaseDir,
 	masterConfigPath,