diff --git a/packages/lib/link/connectors.js b/packages/lib/link/connectors.js index ef78fefd..91a6f995 100644 --- a/packages/lib/link/connectors.js +++ b/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(); diff --git a/packages/master/master.js b/packages/master/master.js index aec2aa05..54ac0933 100755 --- a/packages/master/master.js +++ b/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) { diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js index 47050662..389bd86c 100644 --- a/test/integration/clusterio.js +++ b/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() { diff --git a/test/integration/index.js b/test/integration/index.js index be6eb761..d1588ae9 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -197,6 +197,7 @@ module.exports = { getControl, url, + controlToken, instancesDir, databaseDir, masterConfigPath,