Rewrite reconnection logic

Implement proper session tracking and resume on reconnect for the
WebSocket server.

test/integration/index.js

@@ -30,6 +30,8 @@ async function get(path) {
 let masterProcess;
 let slaveProcess;
 
+let url = "https://localhost:4443/";
+let token = jwt.sign({ id: "api" }, "TestSecretDoNotUse");
 let instancesDir = path.join("test", "temp", "instances");
 let databaseDir = path.join("test", "temp", "databse");
 let masterConfigPath = path.join("test", "temp", "master-integration.json");
@@ -71,16 +73,16 @@ before(async function() {
 	await fs.remove(slaveConfigPath);
 	await fs.remove(controlConfigPath);
 
-	let token = jwt.sign({ id: "api" }, "TestSecretDoNotUse");
 	await exec(`node master --config ${masterConfigPath} config set master.database_directory ${databaseDir}`);
 	await exec(`node master --config ${masterConfigPath} config set master.auth_secret TestSecretDoNotUse`);
 	await exec(`node master --config ${masterConfigPath} config set master.http_port 8880`);
 	await exec(`node master --config ${masterConfigPath} config set master.https_port 4443`);
+	await exec(`node master --config ${masterConfigPath} config set master.heartbeat_interval 0.25`);
 
 	await exec(`node client --config ${slaveConfigPath} config set slave.id 4`);
 	await exec(`node client --config ${slaveConfigPath} config set slave.name slave`);
 	await exec(`node client --config ${slaveConfigPath} config set slave.instances_directory ${instancesDir}`);
-	await exec(`node client --config ${slaveConfigPath} config set slave.master_url "https://localhost:4443/"`);
+	await exec(`node client --config ${slaveConfigPath} config set slave.master_url "${url}"`);
 	await exec(`node client --config ${slaveConfigPath} config set slave.master_token "${token}"`);
 
 	await exec(`node clusterctl --config ${controlConfigPath} control-config set control.master_url "https://localhost:4443/"`);
@@ -111,6 +113,8 @@ module.exports = {
 	get,
 	exec,
 
+	url,
+	token,
 	instancesDir,
 	databaseDir,
 	masterConfigPath,

test/integration/link.js

@@ -0,0 +1,70 @@
+const assert = require("assert").strict;
+const events = require("events");
+
+const link = require("lib/link");
+
+const { get, exec, url, token } = require("./index");
+
+
+class TestControl extends link.Link {
+	constructor(connector) {
+		super("control", "master", connector);
+		link.attachAllMessages(this);
+	}
+
+	async shutdownConnectionRequestHandler() { }
+	async instanceOutputEventHandler() { }
+}
+
+class TestControlConnector extends link.WebSocketClientConnector {
+	register() {
+		this.sendHandshake("register_control", { token: this.token, agent: "clusterctl", version: "test" });
+	}
+}
+
+describe("Integration of lib/link", function() {
+	let control;
+	let controlConnector;
+	it("should emitt an error if authentication failed", async function() {
+		controlConnector = new TestControlConnector(url, 0.2);
+		controlConnector.token = "gibberish";
+		control = new TestControl(controlConnector);
+		await assert.rejects(
+			controlConnector.connect(),
+			new Error("Authentication failed")
+		);
+	});
+
+	it("should connect with proper credentials", async function() {
+		controlConnector = new TestControlConnector(url, 0.2);
+		controlConnector.token = token;
+		control = new TestControl(controlConnector);
+		await controlConnector.connect();
+	});
+
+	it("should reconnect on connection lost", async function() {
+		controlConnector._socket.close(1008, "Test");
+		await events.once(controlConnector, "connected");
+	});
+
+	it("should reconnect on connection terminated", async function() {
+		controlConnector._socket.terminate();
+		await events.once(controlConnector, "connected");
+	});
+
+	it("should reconnect on connection heartbeat timeout", async function() {
+		controlConnector.stopHeartbeat();
+		await events.once(controlConnector, "connected");
+	});
+
+	it("should invalidate on reconnection with bad session token", async function() {
+		controlConnector._sessionToken = "blah";
+		controlConnector.stopHeartbeat();
+		await events.once(controlConnector, "invalidate");
+		await events.once(controlConnector, "connected");
+	});
+
+	after(async function() {
+		controlConnector.close(1001, "Test Quit");
+	});
+});

test/lib/link/connector.js

@@ -6,16 +6,35 @@ const link = require('lib/link');
 
 
 describe("lib/link/connectors", function() {
-	describe("class WebSocketClientConnector", function() {
-		let testConnector = new link.WebSocketClientConnector('source', 'url');
-		testConnector._socket = new mock.MockSocket();
-		describe(".disconnect()", function() {
-			it("should call terminate on the socket", function() {
-				testConnector._socket.terminateCalled = false;
-				testConnector.disconnect();
-				assert(testConnector._socket.terminateCalled, "Close was not called");
+	describe("class WebSocketBaseConnector", function() {
+		let testConnector = new link.WebSocketBaseConnector();
+
+		describe("._dropSendBufferSeq()", function() {
+			beforeEach(function() {
+				testConnector._sendBuffer = [
+					{ seq: 1, type: "test", data: {} },
+					{ seq: 2, type: "test", data: {} },
+					{ seq: 4, type: "test", data: {} },
+				];
+			});
+			it("should do nothing if passed null", function() {
+				testConnector._dropSendBufferSeq(null);
+				assert(testConnector._sendBuffer.length === 3, "entries were dropped");
+			});
+			it("should drop up to sequence", function() {
+				testConnector._dropSendBufferSeq(2);
+				assert.equal(testConnector._sendBuffer.length, 1, "incorrect number of entries dropped");
+			});
+			it("should ignore holes", function() {
+				testConnector._dropSendBufferSeq(3);
+				assert.equal(testConnector._sendBuffer.length, 1, "incorrect number of entries dropped");
 			});
 		});
+	});
+
+	describe("class WebSocketClientConnector", function() {
+		let testConnector = new link.WebSocketClientConnector('url', 1);
+		testConnector._socket = new mock.MockSocket();
 
 		describe(".register()", function() {
 			it("is abstract", function() {
@@ -26,8 +45,21 @@ describe("lib/link/connectors", function() {
 			});
 		});
 
+		describe(".sendHandshake()", function() {
+			it("calls send on the socket", function() {
+				testConnector._state = "handshake";
+				testConnector._socket.sentMessages = [];
+				testConnector.sendHandshake('test', { test: true });
+				assert.deepEqual(
+					testConnector._socket.sentMessages.map(JSON.parse),
+					[{ seq: null, type: 'test', data: { test: true }}]
+				);
+			});
+		});
+
 		describe(".send()", function() {
 			it("calls send on the socket", function() {
+				testConnector._state = "connected";
 				testConnector._socket.sentMessages = [];
 				let seq = testConnector.send('test', { test: true });
 				assert.deepEqual(
@@ -38,43 +70,32 @@ describe("lib/link/connectors", function() {
 		});
 
 		describe(".close()", function() {
-			it("should send close", function() {
-				testConnector._socket.sentMessages = [];
-				let called = false;
+			it("should call close", function() {
+				testConnector._socket.closeCalled = false;
 				testConnector.close("test reason");
-				assert.deepEqual(
-					testConnector._socket.sentMessages.map(JSON.parse),
-					[{ seq: testConnector._seq - 1, type: 'close', data: { reason: "test reason" } }]
-				)
+				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
 			});
 		});
 
 		describe(".processHandshake()", function() {
 			it("should close on invalid message", function() {
-				testConnector._socket.sentMessages = [];
+				testConnector._socket.closeCalled = false;
 				testConnector._processHandshake({ data: "invalid message" }),
-				assert.deepEqual(
-					testConnector._socket.sentMessages.map(JSON.parse),
-					[{seq: testConnector._seq - 1, type: 'close', data: {
-						reason: "Invalid handshake" }
-					}]
-				);
+				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
 			});
 			it("should call register on hello", function() {
 				let called = false;
 				testConnector.register = () => { called = true; };
-				testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }}),
+				testConnector._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }});
 				assert(called, "register was not called");
 			});
-			it("should emit ready on ready", async function() {
-				let result = events.once(testConnector, 'ready');
-				testConnector._processHandshake({ seq: 1, type: 'ready', data: {}}),
+			it("should emit connected on ready", async function() {
+				let result = events.once(testConnector, 'connected');
+				testConnector._processHandshake(
+					{ seq: 1, type: 'ready', data: { session_token: "x", heartbeat_interval: 10 }}
+				);
 				await result;
-			});
-			it("should emit error on close", async function() {
-				let result = events.once(testConnector, 'ready');
-				testConnector._processHandshake({ seq: 1, type: 'close', data: { reason: "test" }}),
-				await assert.rejects(result, new Error("server closed during handshake: test"));
+				clearInterval(testConnector._heartbeatId);
 			});
 		});
 
@@ -97,9 +118,9 @@ describe("lib/link/connectors", function() {
 				testConnector._socket.events.get("message")("{}");
 				assert(called, "_processHandshake was not called");
 			});
-			it("should emit message on message in ready state", async function() {
+			it("should emit message on message in connected state", async function() {
 				testConnector._socket.sentMessages = [];
-				testConnector._state = "ready";
+				testConnector._state = "connected";
 				let result = events.once(testConnector, 'message');
 				testConnector._socket.events.get("message")('{"message":true}');
 				assert.deepEqual(await result, [{ message: true }]);

test/lib/link/link.js

@@ -27,25 +27,6 @@ describe("lib/link/link", function() {
 			}));
 		});
 
-		it("should close connector on invalid messages", function() {
-			testConnector.sentMessages = [];
-			testConnector.emit('message', { invalid: true });
-			assert.deepEqual(
-				testConnector.sentMessages,
-				[{
-					seq: testConnector._seq - 1,
-					type: 'close',
-					data: { reason: "Invalid message: Malformed" },
-				}]
-			);
-		});
-
-		it("should disconnect on close", function() {
-			testConnector.disconnectCalled = false;
-			testConnector.emit('message', { seq: 1, type: 'close', data: { reason: "test" }});
-			assert(testConnector.disconnectCalled, "disconnect was not called");
-		});
-
 		it("should emitt error processing message to connector", async function() {
 			testLink.setHandler('throws', () => { throw new Error("Handler error"); }, () => true);
 			let result = events.once(testConnector, 'message');

test/master.js

@@ -5,14 +5,4 @@ const master = require('../master');
 
 
 describe('Master testing', function() {
-	describe("class WebSocketServerConnector", function() {
-		let testConnector = new master._WebSocketServerConnector(new mock.MockSocket());
-		describe(".disconnect()", function() {
-			it("should call disconnect on the socket", function() {
-				testConnector._socket.terminateCalled = false;
-				testConnector.disconnect();
-				assert(testConnector._socket.terminateCalled, "Terminate was not called");
-			});
-		});
-	});
 });

test/mock.js

@@ -42,15 +42,6 @@ class MockConnector extends events.EventEmitter {
 		setImmediate(() => this.emit('send', message));
 		return this._seq++;
 	}
-
-	close(reason) {
-		this.send('close', { reason });
-		this.disconnect();
-	}
-
-	disconnect() {
-		this.disconnectCalled = true;
-	}
 }
 
 class MockServer {