diff --git a/CHANGELOG.md b/CHANGELOG.md
index f2835276..53365239 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -29,6 +29,8 @@ Version 2.0.0
### Features
- Added export of pollution statitics.
+- Reconnection logic that esures no data is dropped talking to the master
+ server provided the session can be resumed.
### Changes
diff --git a/client.js b/client.js
index 24e45ca8..91b960ab 100644
--- a/client.js
+++ b/client.js
@@ -1,6 +1,7 @@
const fs = require('fs-extra');
const path = require('path');
const yargs = require("yargs");
+const events = require("events");
const version = require("./package").version;
// internal libraries
@@ -402,21 +403,18 @@ class InstanceConnection extends link.Link {
class SlaveConnector extends link.WebSocketClientConnector {
constructor(slaveConfig) {
- super(slaveConfig.get("slave.master_url"));
-
- this.id = slaveConfig.get("slave.id");
- this.name = slaveConfig.get("slave.name");
- this._token = slaveConfig.get("slave.master_token");
+ super(slaveConfig.get("slave.master_url"), slaveConfig.get("slave.reconnect_delay"));
+ this.slaveConfig = slaveConfig;
}
register() {
console.log("SOCKET | registering slave");
- this.send('register_slave', {
- token: this._token,
- agent: 'Clusterio Slave',
+ this.sendHandshake("register_slave", {
+ token: this.slaveConfig.get("slave.master_token"),
+ agent: "Clusterio Slave",
version,
- id: this.id,
- name: this.name,
+ id: this.slaveConfig.get("slave.id"),
+ name: this.slaveConfig.get("slave.name"),
});
}
}
@@ -649,11 +647,13 @@ class Slave extends link.Link {
await this.updateInstances();
}
- async stop() {
+ async shutdown() {
+ await link.messages.shutdownConnection.send(this);
for (let instanceId of this.instanceConnections.keys()) {
await this.stopInstance(instanceId);
}
- this.connector.close("shutdown");
+ this.connector.close(1001, "Slave Shutdown");
+ await events.once(this.connector, "close");
}
}
@@ -851,7 +851,7 @@ async function startClient() {
secondSigint = true;
console.log("Caught interrupt signal, shutting down");
- slave.stop().catch(err => {
+ slave.shutdown().catch(err => {
console.error(`
+---------------------------------------------------------------+
| Unexpected error occured while stopping slave, please report |
diff --git a/clusterctl.js b/clusterctl.js
index edf8baef..89a0509a 100644
--- a/clusterctl.js
+++ b/clusterctl.js
@@ -4,6 +4,7 @@ const yargs = require("yargs");
const version = require("./package").version;
const asTable = require("as-table").configure({delimiter: ' | '});
const chalk = require("chalk");
+const events = require("events");
const link = require("lib/link");
const errors = require("lib/errors");
@@ -315,16 +316,16 @@ commands = new Map([...commands.map(command => [command.name, command])]);
* Connector for control connection to master server
*/
class ControlConnector extends link.WebSocketClientConnector {
- constructor(url, token) {
- super(url);
+ constructor(url, reconnectDelay, token) {
+ super(url, reconnectDelay);
this._token = token;
}
register() {
console.log("SOCKET | registering control");
- this.send('register_control', {
+ this.sendHandshake("register_control", {
token: this._token,
- agent: 'clusterctl',
+ agent: "clusterctl",
version: version,
});
}
@@ -348,6 +349,12 @@ class Control extends link.Link {
let { instance_id, output } = message.data;
console.log(formatOutputColored(output));
}
+
+ async shutdown() {
+ await link.messages.shutdownConnection.send(this);
+ this.connector.close(1001, "Control Quit")
+ await events.once(this.connector, "close");
+ }
}
async function findCredentials(controlConfig) {
@@ -411,6 +418,13 @@ ControlGroup.define({
type: "string",
optional: true,
});
+ControlGroup.define({
+ name: "reconnect_delay",
+ title: "Reconnect Delay",
+ description: "Maximum delay to wait before attempting to reconnect WebSocket",
+ type: "number",
+ initial_value: 2,
+});
ControlGroup.finalize();
class ControlConfig extends config.Config { }
@@ -471,10 +485,19 @@ async function startControl() {
}
let controlConnector = new ControlConnector(
- controlConfig.get("control.master_url"), controlConfig.get("control.master_token")
+ controlConfig.get("control.master_url"),
+ controlConfig.get("control.reconnect_delay"),
+ controlConfig.get("control.master_token")
);
let control = new Control(controlConnector);
- await controlConnector.connect();
+ try {
+ await controlConnector.connect();
+ } catch(err) {
+ if (err instanceof errors.AuthenticationFailed) {
+ throw new errors.StartupError(err.message);
+ }
+ throw err;
+ }
if (commands.has(commandName)) {
command = commands.get(commandName);
@@ -492,13 +515,12 @@ async function startControl() {
process.exitCode = 1;
} else {
- controlConnector.close("error");
throw err;
}
- }
- controlConnector.close("quit");
- return; // ??
+ } finally {
+ await control.shutdown();
+ }
}
//XXX control.close("done");
diff --git a/docs/assets/connection_flow.dot b/docs/assets/connection_flow.dot
new file mode 100644
index 00000000..a8d9643c
--- /dev/null
+++ b/docs/assets/connection_flow.dot
@@ -0,0 +1,20 @@
+digraph "Client Connection Flow" {
+ node [shape=box];
+ graph [newrank=true];
+
+ start [label="Start"];
+ wait_hello [label="Wait for hello"];
+ new_session [label="New session"];
+ wait_ready [label="Wait for ready"];
+ connected [label="Connected"];
+ abort [label="Abort"];
+ lost_connection [label="Connection lost or\nunexpectedly closed\nwhile connecting" style="dotted"];
+
+
+ lost_connection -> start [label="Wait random delay"];
+ start -> wait_hello [label="Connect to /api/socket"];
+ wait_hello -> new_session [label="Receive hello"];
+ new_session -> wait_ready [label="Send register_*"];
+ wait_ready -> connected [label="Receive ready"];
+ wait_ready -> abort [label="Connection closed with 4003"];
+}
diff --git a/docs/assets/connection_flow.svg b/docs/assets/connection_flow.svg
new file mode 100644
index 00000000..95414122
--- /dev/null
+++ b/docs/assets/connection_flow.svg
@@ -0,0 +1,99 @@
+
+
+
+
+
diff --git a/docs/assets/generate.sh b/docs/assets/generate.sh
new file mode 100755
index 00000000..6c90f7d5
--- /dev/null
+++ b/docs/assets/generate.sh
@@ -0,0 +1,4 @@
+for x in $(dirname $0)/*.dot; do
+ echo $x
+ dot $x -Tsvg -o ${x/.dot/.svg}
+done
diff --git a/docs/assets/reconnection_flow.dot b/docs/assets/reconnection_flow.dot
new file mode 100644
index 00000000..19324276
--- /dev/null
+++ b/docs/assets/reconnection_flow.dot
@@ -0,0 +1,23 @@
+digraph "Client Reconnection Flow" {
+ node [shape=box];
+ graph [newrank=true];
+
+ reconnect [label="Reconnect"];
+ wait_hello [label="Wait for hello"];
+ resume_session [label="Resume session"];
+ new_session [label="New session" style="dashed"];
+ wait_continue [label="Wait for continue"];
+ connected [label="Connected"];
+ lost_connection [label="Connection lost or\nunexpectedly closed\nwhile reconnecting" style="dotted"];
+ abort [label="Abort"];
+
+ { rank="same"; wait_continue; new_session };
+
+ lost_connection -> reconnect [label="Wait random delay"];
+ reconnect -> wait_hello [label="Connect to /api/socket"];
+ wait_hello -> resume_session [label="Receive hello"];
+ resume_session -> wait_continue [label="Send resume"];
+ wait_continue -> abort [label="Connection closed with 4003"];
+ wait_continue -> new_session [label="Receive invalidate"];
+ wait_continue -> connected [label="Receive ready"];
+}
diff --git a/docs/assets/reconnection_flow.svg b/docs/assets/reconnection_flow.svg
new file mode 100644
index 00000000..22c7a4b4
--- /dev/null
+++ b/docs/assets/reconnection_flow.svg
@@ -0,0 +1,112 @@
+
+
+
+
+
diff --git a/docs/assets/server_flow.dot b/docs/assets/server_flow.dot
new file mode 100644
index 00000000..c16e6db0
--- /dev/null
+++ b/docs/assets/server_flow.dot
@@ -0,0 +1,21 @@
+digraph "Server Connection Flow" {
+ node [shape=box];
+ graph [newrank=true];
+
+ new_connection [label="New connection"];
+ wait_handshake [label="Wait for handshake"];
+ validate_register [label="Validate register"];
+ validate_resume [label="Validate resume"];
+ connected [label="Connected"];
+ close [label="Close connection"];
+
+ new_connection -> wait_handshake [label="Send hello"];
+ wait_handshake -> validate_register [label="Receive register_*"];
+ wait_handshake -> validate_resume [label="Receive resume"];
+ wait_handshake -> close [label="Bad handshake\nTimeout"];
+ validate_register -> close [label="Bad token\nBad register"];
+ validate_register -> connected [label="Valid register:\nSend ready"];
+ validate_resume -> close [label="Bad token\nBad resume"];
+ validate_resume -> wait_handshake [label="Bad session:\nsend invalidate"];
+ validate_resume -> connected [label="Valid resume:\nSend continue"];
+}
diff --git a/docs/assets/server_flow.svg b/docs/assets/server_flow.svg
new file mode 100644
index 00000000..3dedab60
--- /dev/null
+++ b/docs/assets/server_flow.svg
@@ -0,0 +1,118 @@
+
+
+
+
+
diff --git a/docs/socket.md b/docs/socket.md
index 790e7b0d..597ef535 100644
--- a/docs/socket.md
+++ b/docs/socket.md
@@ -1,4 +1,4 @@
-Socket.io protocol
+WebSocket protocol
==================
This document describes the protocol used between the master and
@@ -7,64 +7,211 @@ to be very useful for you.
The protocol is built on WebSocket and uses JSON encoded text messages
to communicate to and from the master server. These messages contain
-the following three properties:
-- seq - integer - The sequence number of the message
+a JSON encoded object with the following three properties:
+
+- seq - integer|null - The sequence number of the message
- type - string - The type of message
- data - object - Message data
+For the handshake and heartbeats the seq property is null.
+
Handshake
---------
Upon connecting to the WebSocket endpoint the server will respond with a
`hello` message containing the master server version as the data. The
-client is then expected to send a `register_slave` or `register_control`
-message back depending on what kind of client it is. If the register
-operation is successful the master will reply with a `ready` message
-with an empty data payload.
+client is then expected to send either `register_*` or a `resume`
+message to set up the connection with the master. The exact flow is
+described below.
-**Master `hello` data**
+
+### Client connection flow
+
+For a client making a new connecting to the master server the first
+course of action after obtaining a token is to connect with WebSocket to
+the /api/socket endpoint and wait for the server hello. After the
+hello, the client sends a register\_\* message corresponding to the type
+of the client. If successful the server will reply with a ready message
+and the connection has been established. See the flowchart below.
+
+
+
+If for any reason the connection is closed or lost during the connection
+attempt the process is started over after a random delay, with the
+exception of the connection being closed with a 4003 code which means
+authentication failed and the client should stop trying to connect. The
+wait for hello and wait for ready states should also have a timeout,
+after which the connection is closed if the hello/reply wasn't received
+in time.
+
+
+### Client reconnection flow
+
+If the connection is lost after the client has received the ready
+message from the server it should attempt a reconnect instead. This is
+very similar to a new connection, but instead of a register\_\* messsage
+the client sends a resume message and the server replies with a continue
+message if resuming the session was successful. This continue message
+has sequence number of the last message the master received and the
+client is expected to imediately send all messages from its send buffer
+that occured after the sequence number given.
+
+The server will respond with an invalidate message if resuming the
+session is not possible. In this case the client should continue from
+the new session state of the connection flow (see the previous section).
+
+
+
+Like the connection flow the process starts over after a delay if the
+connection is closed or lost. The same exception for connection closed
+with 4003 applies as with the connection flow. Additionally after an
+invalidate message has been received and the connection flow fails, it
+should start over from the connection flow.
+
+
+### Server connection flow
+
+From the master server side new connections and reconnection attempts
+start out the same, a new connection has been made and the server sends
+the hello messoge to the client. If a vaild register\_\* message is
+received a new connection is set up for the client and the ready message
+is sent in reply. If a valid resume message is received and the session
+that's attempted to be resumed is active it'll signal that the session
+is resumed with a continue message and the client upon receiving the
+continue message should send the messages from its send buffer from
+after the last\_seq given in the continue message.
+
+If the session token used to resume is not valid it'll send an invalidate
+message and expect the client to continue with a registr\_\* message. If an
+invalid or unexpected message is received during the handshake the connection
+is closed. See the flowchart below.
+
+
+
+For authentication failure the connection is closed with code 4003,
+which the connection client should interpret as connecting will never
+succeed and stop trying to reconnect.
+
+
+### Connection tracking and heartbeats
+
+Once connected as indicated by the ready and continue messages from the
+master server the client must start sending heartbeats at the interval
+given in the ready/continue message, and storing messages sent over the
+link in a sending side buffer. The heartbeats sent must have the seq of
+the last received message on the link and acts as an ack for received
+messages. The master server also sends heartbeats in the same manner
+and the client should remove messages up to and including the sequence
+number received in the heartbeat from its send buffer.
+
+
+Handshake messages
+------------------
+
+### `hello`
+
+Sent by the master server when the WebSocket connection has been opened.
- version - string - The version of the master, e.g. "2.0.0".
-**Slave `register_slave` response data**
+### `register_slave`
-- agent - string - Human readable identifying the software connecting.
+Slave handshake for establishing a new connection session.
+
+- token - string - Authentication token.
+- agent - string - Human readable string of the software connecting.
- version - string - The protocol version of the slave, e.g. "2.0.0".
- name - string - Name of the slave.
- id - integer - ID of the slave.
-**Control `register_control repsonse data**
+### `register_control`
-- agent - string - Human readable identifying the software connecting.
+Control handshake for enstablishing a new connection session.
+
+- token - string - Authentication token.
+- agent - string - Human readable string of the software connecting.
- version - string - The protocol version of the control, e.g. "2.0.0".
-**Example handshake**
-```js
-// Server hello
-({
- type: 'hello',
- data: {
- version: "2.0.0",
- }
-})
+### `resume`
-// Client response
-({
- type: 'register_slave',
- data: {
- agent: 'Clusterio Slave',
- version: "2.0.0",
- name: "Foo",
- id: 123,
- }
-})
+Handshake for resuming an existing connection session where the
+connection had dropped unexpectedly.
-// Server repsonse
-({
- type: 'ready',
- data: {}
-})
+- session_token - string - Session token to resume session with.
+- last_seq - integer|null - Last message seq received by the client.
+
+### `invalidate`
+
+Sent by the master server in reply to a resume for a session that cannot
+be resumed. The client should start a new session with a register\_\*
+message in response.
+
+### `continue`
+
+Sent by the master server to signal resuming the session was successful.
+
+- last_seq - integer|null - Last message seq received by the server.
+- heartbeat_interval - number - Rate client should send heartbeats at.
+
+### `ready`
+
+Sent by the master server to signal the connection has been established
+with a new session.
+
+- session_token - string - Token to resume the session with.
+- heartbeat_interval - number - Rate client should send heartbeats at.
+
+### `heartbeat`
+
+Sent by both the master server and clients at the interval given in the
+continue/ready messages.
+
+- last_seq - integer - Last message seq received over the link.
+
+
+Example handshake
+-----------------
+
+Server hello after opening the WebSocket.
+
+```json
+{
+ "seq": null,
+ "type": "hello",
+ "data": {
+ "version": "2.0.0"
+ }
+}
+```
+
+Slave registering a new connection session.
+
+```json
+{
+ "seq": null,
+ "type": "register_slave",
+ "data": {
+ "token": "xyzabc123",
+ "agent": "Clusterio Slave",
+ "version": "2.0.0",
+ "name": "Foo",
+ "id": 123
+ }
+}
+```
+
+Master server reply signaling the session is established and ready.
+
+```json
+{
+ "seq": null,
+ "type": "ready",
+ "data": {
+ "session_token": "xyz.azy.bzy",
+ "heartbeat_interval": 45
+ }
+}
```
@@ -73,11 +220,11 @@ Events
Events are messages send in one direction over the link, that the
receiver is expected to act upon, but not make any replies (for that
-there's [Requests](#requests). By convention a message types ending in
-`*_event` is an event. The data payload of an event consists solely of
+there's [Requests](#requests). By convention message types ending in
+`*_event` are events. The data payload of an event consists solely of
event specific properties.
-See [lib/link/messages.js](lib/link/messages.js) for the recognized
+See [lib/link/messages.js](/lib/link/messages.js) for the recognized
events and their contents.
@@ -87,30 +234,47 @@ Requests
Requests are messages which expects a response in return. They function
a lot like HTTP requests except that both parties of a connection can
inniate a request that is expected to be responded to. By convention
-message types ending in `*_request` is a request, and is expected to be
+message types ending in `*_request` are requests, and is expected to be
replied to with a corresponding `*_response` message. If an error
occured while processing the request an error response containing an
error property in the data payload is sent instead.
-**`*_response` data**
+See [lib/link/messages.js](/lib/link/messages.js) for the recognized
+requests and their contents.
+
+### `*_response`
- seq - integer - The seq number of the request message.
- additional properties may be present depending on the type of request.
-**`*_resopnse` error data**
+### `*_response` error
+
- seq - integer - The seq number of the request message.
- error - string - Human readable text message describing the error.
-See [lib/link/messages.js](lib/link/messages.js) for the recognized
-requests and their contents.
+Graceful shutdown
+-----------------
-Closing the connection
-----------------------
+To ensure no data is lost when closing the link it's important to
+send connection shutdown request and wait for it to complete before
+closing the WebSocket. Either party of the link can send this request
+and the party sending the request should be the one that initates the
+close of the WebSocket.
-The party that innitates the shut down of the connection should send a
-close message containing the reason for closing before disconnecting.
+### `connection_shutdown_request`
-**`close` data**
+Request graceful shutdown of the connection. The receiving party should
+stop tasks that send events and requests over the link and wait for
+pending requests it has over the link to complete, but it should not
+stop responding to requests sent to it. Only after this process is done
+should the response be sent.
-- reason - string - Reason for closing the connection
+### `connection_shutdown_response`
+
+Signals the shutdown is complete and that the connection can be closed.
+The party that initiatied the shutdown request may still send other
+requests/events over the link to finish its part of the shutdown.
+
+If the error response variant is received it should be logged if
+possible and treated as a success.
diff --git a/lib/config/definitions.js b/lib/config/definitions.js
index b1fdcde1..b52c7603 100644
--- a/lib/config/definitions.js
+++ b/lib/config/definitions.js
@@ -73,6 +73,13 @@ MasterGroup.define({
return bytes.toString("base64");
},
});
+MasterGroup.define({
+ name: "heartbeat_interval",
+ title: "Heartbeat Interval",
+ description: "Interval heartbeats are sent out on WebSocket connections",
+ type: "number",
+ initial_value: 60,
+});
MasterGroup.finalize();
class MasterConfig extends classes.Config { }
@@ -123,6 +130,13 @@ SlaveGroup.define({
type: "string",
initial_value: "localhost",
});
+SlaveGroup.define({
+ name: "reconnect_delay",
+ title: "Reconnect Delay",
+ description: "Maximum delay to wait before attempting to reconnect WebSocket",
+ type: "number",
+ initial_value: 5,
+});
SlaveGroup.finalize();
class SlaveConfig extends classes.Config { }
diff --git a/lib/errors.js b/lib/errors.js
index 2da6fbef..2945d13d 100644
--- a/lib/errors.js
+++ b/lib/errors.js
@@ -14,6 +14,9 @@ class InvalidMessage extends Error {
}
}
+// Thrown when WebSocket authentication failed
+class AuthenticationFailed extends Error { }
+
// Errror class for known errors occuring during startup
class StartupError extends Error { }
@@ -33,6 +36,7 @@ module.exports = {
CommandError,
RequestError,
InvalidMessage,
+ AuthenticationFailed,
StartupError,
EnvironmentError,
PluginError,
diff --git a/lib/link/connectors.js b/lib/link/connectors.js
index 6311ce2e..55a33436 100644
--- a/lib/link/connectors.js
+++ b/lib/link/connectors.js
@@ -3,19 +3,122 @@ const events = require("events");
const WebSocket = require("ws");
const schema = require("lib/schema");
+const errors = require("lib/errors");
+class WebSocketBaseConnector extends events.EventEmitter {
+ constructor() {
+ super();
+
+ this._socket = null;
+ this._seq = 1;
+ this._lastHeartbeat = null;
+ this._heartbeatId = null;
+ this._heartbeatInterval = null;
+ this._lastReceivedSeq = null;
+ this._sendBuffer = [];
+ }
+
+ _dropSendBufferSeq(seq) {
+ if (seq === null) {
+ return;
+ }
+
+ let dropCount = 0;
+ for (let index = 0; index < this._sendBuffer.length; index++) {
+ if (this._sendBuffer[index].seq <= seq) {
+ dropCount++;
+ } else {
+ break;
+ }
+ }
+ this._sendBuffer.splice(0, dropCount);
+ }
+
+ _doHeartbeat() {
+ this._check(["connected"]);
+ if (Date.now() - this._lastHeartbeat > 2000 * this._heartbeatInterval) {
+ console.log("SOCKET | closing after heartbeat timed out");
+ this._socket.close(1008, "Heartbeat timeout");
+ clearInterval(this._heartbeatId);
+ this._heartbeatId = null;
+
+ } else {
+ this._socket.send(JSON.stringify({
+ seq: null,
+ type: "heartbeat",
+ data: { seq: this._lastReceivedSeq },
+ }));
+ }
+ }
+
+ _processHeartbeat(message) {
+ if (!schema.heartbeat(message)) {
+ console.log("SOCKET | closing after received invalid heartbeat:", message);
+ this._socket.close(1002, "Invalid heartbeat");
+ console.log(schema.heartbeat.errors);
+ return;
+ }
+
+ this._lastHeartbeat = Date.now();
+ this._dropSendBufferSeq(message.data.seq);
+ }
+
+ /**
+ * Start sending heartbeats
+ */
+ startHeartbeat() {
+ this._lastHeartbeat = Date.now();
+ this._heartbeatId = setInterval(() => {
+ this._doHeartbeat();
+ }, this._heartbeatInterval * 1000);
+ }
+
+ /**
+ * Stop sending heartbeats
+ */
+ stopHeartbeat() {
+ if (this._heartbeatId) {
+ clearInterval(this._heartbeatId);
+ this._heartbeatId = null;
+ }
+ }
+
+ /**
+ * Send a message over the socket
+ *
+ * @returns the sequence number of the message sent
+ */
+ send(type, data) {
+ this._check(["handshake", "connected"]);
+ let message = { seq: this._seq, type, data };
+ this._sendBuffer.push(message);
+ if (this._state === "connected") {
+ this._socket.send(JSON.stringify(message));
+ }
+ return this._seq++;
+ }
+}
+
/**
* Connector for master server clients
*/
-class WebSocketClientConnector extends events.EventEmitter {
- constructor(url) {
+class WebSocketClientConnector extends WebSocketBaseConnector {
+ constructor(url, reconnectDelay) {
super();
this._url = url;
- this._seq = 1;
+ // The following states are used in the client connector
+ // new: Not connected
+ // handshake: Attempting to (re)connect to server.
+ // connected: Connection is online
+ // closing: Connection is in the process of being closed.
this._state = "new";
+ this._reconnectId = null;
+ this._sessionToken = null;
+ this._connected = false;
+ this._reconnectDelay = reconnectDelay;
}
_check(expectedStates) {
@@ -27,23 +130,38 @@ class WebSocketClientConnector extends events.EventEmitter {
}
/**
- * Send a message over the socket
+ * Send a handshake message over the socket
*
- * @returns the sequence number of the message sent
+ * Used in the register function in order to send the register handshake
+ * message over the WebSocket.
*/
- send(type, data) {
- this._socket.send(JSON.stringify({ seq: this._seq, type, data }));
- return this._seq++;
+ sendHandshake(type, data) {
+ this._check(["handshake"]);
+ this._socket.send(JSON.stringify({ seq: null, type, data }));
}
/**
* Close the connection with the given reason.
*
- * Sends a close message and disconnects the connector.
+ * Sends a close frame and disconnects the connector.
*/
- close(reason) {
- this.send('close', { reason });
- this._socket.close();
+ close(code, reason) {
+ if (this._state === "new") {
+ return;
+ }
+
+ if (this._reconnectId) {
+ clearTimeout(this._reconnectId);
+ this._reconnectId = null;
+ }
+
+ this.stopHeartbeat();
+
+ if (["handshake", "connected"].includes(this._state)) {
+ this._socket.close(code, reason);
+ }
+
+ this._state = "closing";
}
/**
@@ -53,6 +171,13 @@ class WebSocketClientConnector extends events.EventEmitter {
this._check(["new"]);
this._state = "handshake";
+ this._doConnect();
+
+ // Wait for the connection to be ready
+ await events.once(this, "connected");
+ }
+
+ async _doConnect() {
let url = new URL(this._url);
url.pathname += "api/socket";
@@ -66,27 +191,73 @@ class WebSocketClientConnector extends events.EventEmitter {
});
this._attachSocketHandlers();
+ }
- // Wait for the connection to be ready
- await events.once(this, 'ready');
+ /**
+ * Attempt re-establishing the connecting again
+ */
+ reconnect() {
+ if (this._reconnectId !== null) {
+ return;
+ }
+
+ let delay = Math.random() * this._reconnectDelay;
+ console.log(`SOCKET | waiting ${delay.toLocaleString("en", { maximumFractionDigits: 2 })} seconds for reconnect`)
+ this._reconnectId = setTimeout(() => {
+ this._reconnectId = null;
+ this._doConnect();
+ }, delay * 1000);
}
_attachSocketHandlers() {
this._socket.on("close", (code, reason) => {
console.log(`SOCKET | Close (code: ${code}, reason: ${reason})`);
+ // Authentication failed
+ if (code === 4003) {
+ this.emit("error", new errors.AuthenticationFailed("Authentication failed"));
+ this._state = "closing";
+ }
+
+ if (this._state === "closing") {
+ this._lastReceivedSeq = null;
+ this._sessionToken = null;
+ this._sendBuffer.length = 0;
+ this._state = "new";
+
+ if (this._connected) {
+ this.emit("close");
+ this._connected = false;
+ }
+
+ } else {
+ this._state = "handshake";
+ this.reconnect();
+
+ if (this._connected) {
+ this.emit("dropped");
+ this._connected = false;
+ }
+ }
+
+ this.stopHeartbeat();
});
+
this._socket.on("error", err => {
+ // It's assumed that close is always called by ws
console.error("SOCKET | Error:", err);
- this._socket.close();
});
+
+ /*
this._socket.on("unexpected-response", (req, res) => {
let data = "";
res.on('data', (chunk) => { data += chunk; });
res.on('end', () => {
console.error(`SOCKET | Unexpeced Response ${res.statusCode}`);
console.error(data);
+ this._socket.emit("error", new Error(`Unexpected Response ${res.statusCode}`));
});
});
+ */
this._socket.on("open", () => {
console.log("SOCKET | Open");
@@ -103,42 +274,73 @@ class WebSocketClientConnector extends events.EventEmitter {
let message = JSON.parse(data);
if (this._state === "handshake") {
this._processHandshake(message);
- } else if (this._state === "ready") {
- this.emit("message", message);
+
+ } else if (["connected", "closing"].includes(this._state)) {
+ if (message.seq !== null) {
+ this._lastReceivedSeq = message.seq;
+ }
+
+ if (message.type === "heartbeat") {
+ this._processHeartbeat(message);
+
+ } else {
+ this.emit("message", message);
+ }
+
} else {
throw new Error(`Received message in unexpected state ${this._state}`);
}
});
}
- /**
- * Immediatly close the connection
- */
- disconnect() {
- this._socket.terminate();
- }
-
_processHandshake(message) {
if (!schema.serverHandshake(message)) {
console.log("SOCKET | closing after received invalid handshake:", message);
- this.close("Invalid handshake");
+ this._socket.close(1002, "Invalid handshake");
return;
}
let { seq, type, data } = message;
if (type == 'hello') {
console.log(`SOCKET | received hello from master version ${data.version}`);
- this.register();
+ if (this._sessionToken) {
+ console.log("SOCKET | Attempting resume");
+ this.sendHandshake("resume", {
+ session_token: this._sessionToken,
+ last_seq: this._lastReceivedSeq,
+ });
+ } else {
+ this.register();
+ }
} else if (type === 'ready') {
console.log("SOCKET | received ready from master");
- this._state = "ready";
- this.emit('ready');
+ this._state = "connected";
+ this._sessionToken = data.session_token;
+ this._heartbeatInterval = data.heartbeat_interval;
+ this.startHeartbeat();
+ this._connected = true;
+ this.emit("connected");
- } else if (type === 'close') {
- console.log(`SOCKET | received close from server: ${data.reason}`);
- this.emit('error', new Error(`server closed during handshake: ${data.reason}`));
- this._socket.terminate();
+ } else if (type === "continue") {
+ console.log("SOCKET | resuming existing session");
+ this._state = "connected";
+ this._heartbeatInterval = data.heartbeat_interval;
+ this.startHeartbeat();
+ this._dropSendBufferSeq(data.last_seq);
+ for (let message of this._sendBuffer) {
+ this._socket.send(JSON.stringify(message));
+ }
+ this._connected = true;
+ this.emit("connected");
+
+ } else if (type === "invalidate") {
+ this._heartbeatInterval = null;
+ this._lastReceivedSeq = null;
+ this._sessionToken = null;
+ this._sendBuffer.length = 0;
+ this.emit("invalidate");
+ this.register();
}
}
@@ -186,27 +388,10 @@ class VirtualConnector extends events.EventEmitter {
this.other.emit('message', { seq: this._seq, type, data });
return this._seq++;
}
-
- /**
- * Close the connection with the given reason.
- *
- * Sends a close message to the other end of the connector with the
- * given reason. Doesn't actually disconnect as virtual connectors
- * doen't have a concept of disconnecting.
- */
- close(reason) {
- this.send('close', { reason });
- }
-
- /**
- * Does nothing
- *
- * Virtual connectors do not have a concept of disconnecting.
- */
- disconnect() { }
}
module.exports = {
+ WebSocketBaseConnector,
WebSocketClientConnector,
VirtualConnector,
};
diff --git a/lib/link/link.js b/lib/link/link.js
index 6907d708..f387f8f8 100644
--- a/lib/link/link.js
+++ b/lib/link/link.js
@@ -32,7 +32,6 @@ class Link {
this.processMessage(payload);
} catch (err) {
if (err instanceof errors.InvalidMessage) {
- this.connector.close(`Invalid message: ${err.message}`);
console.error(`Invalid message on ${this.source}-${this.target} link: ${err.message}`)
if (err.errors) {
console.error(err.errors);
@@ -43,10 +42,15 @@ class Link {
}
});
- this.setHandler('close', message => {
- console.log(`SOCKET | Received close: ${message.data.reason}`);
- this.connector.disconnect();
- }, schema.close);
+ connector.on("invalidate", () => {
+ for (let waiterType of this._waiters.values()) {
+ for (let waiter of waiterType) {
+ waiter.reject(new Error("Session lost"));
+ }
+ }
+
+ this._waiters.clear();
+ });
}
/**
@@ -185,13 +189,41 @@ class Link {
throw new Error(`No validator for ${type} on ${this.source}-${this.target}`);
}
- return new Promise((resolve, reject) => {
+ let waiter = { data };
+ waiter.promise = new Promise((resolve, reject) => {
+ waiter.resolve = resolve;
+ waiter.reject = reject;
if (!this._waiters.has(type)) {
- this._waiters.set(type, [{ data, resolve }]);
+ this._waiters.set(type, [waiter]);
} else {
- this._waiters.get(type).push({ data, resolve });
+ this._waiters.get(type).push(waiter);
}
});
+ return waiter.promise;
+ }
+
+ /**
+ * Handle ping requests
+ */
+ async pingRequestHandler() { }
+
+ /**
+ * Prepare connection for close
+ *
+ * Waits for all pending requests on the connection to resolve.
+ * Sub-classes should implement this handler with code that prevents
+ * additional requests to be sent out before calling the super class
+ * method.
+ */
+ async shutdownConnectionRequestHandler() {
+ let promises = [];
+ for (let waiterType of this._waiters.values()) {
+ for (let waiter of waiterType) {
+ promises.push(waiter.promise.then(() => {}, () => {}));
+ }
+ }
+
+ await Promise.all(promises);
}
}
diff --git a/lib/link/messages.js b/lib/link/messages.js
index 7cd22e8d..9108cffb 100644
--- a/lib/link/messages.js
+++ b/lib/link/messages.js
@@ -183,6 +183,18 @@ class Request {
let messages = {}
+// Connection requests
+let wsLinks = ["master-control", "control-master", "master-slave", "slave-master"];
+messages.shutdownConnection = new Request({
+ type: "shutdown_connection",
+ links: wsLinks,
+});
+
+messages.ping = new Request({
+ type: "ping",
+ links: wsLinks,
+});
+
// Management requests
messages.listSlaves = new Request({
type: 'list_slaves',
diff --git a/lib/schema.js b/lib/schema.js
index 788dd339..fec79da6 100644
--- a/lib/schema.js
+++ b/lib/schema.js
@@ -4,27 +4,9 @@ const ajv = new (require('ajv'))({
extendRefs: "fail",
strictDefaluts: true,
strictKeywords: true,
-})
+});
-const closeSchema = {
- $schema: "http://json-schema.org/draft-07/schema#",
- title: "Close Message",
- description: "Format for a close messages on a link",
- properties: {
- "type": { const: "close" },
- "data": {
- additionalProperties: false,
- required: ["reason"],
- properties: {
- "reason": { type: "string" },
- }
- },
- },
-}
-
-const close = ajv.compile(closeSchema);
-
/**
* The base format for any message sent over the link
*/
@@ -36,28 +18,34 @@ const messageSchema = {
additionalProperties: false,
required: ["seq", "type", "data"],
properties: {
- "seq": { type: "integer" },
+ "seq": { type: ["integer", "null"] },
"type": { type: "string" },
"data": { type: "object" },
},
-}
+};
const message = ajv.compile(messageSchema);
-const serverActions = [
- {},
-];
-
-const actionResponse = {
+const heartbeat = ajv.compile({
+ $schema: "http://json-schema.org/draft-07/schema#",
+ title: "Heartbeat",
+ description: "Heartbeat sent over link",
+ type: "object",
additionalProperties: false,
- required: ["action", "id", "status", "message"],
+ required: ["seq", "type", "data"],
properties: {
- "action": { enum: ["TODO"] },
- "id": { type: "integer" },
- "status": { enum: ["started", "done", "error"] },
- "message": { type: "string" },
+ "seq": { const: null },
+ "type": { const: "heartbeat" },
+ "data": {
+ type: "object",
+ additionalProperties: false,
+ required: ["seq"],
+ properties: {
+ "seq": { type: ["integer", "null"] },
+ },
+ },
},
-};
+});
const serverHandshake = ajv.compile({
$schema: "http://json-schema.org/draft-07/schema#",
@@ -82,15 +70,33 @@ const serverHandshake = ajv.compile({
"type": { const: "ready" },
"data": {
additionalProperties: false,
- /* XXX anything here?
- required: ["version"],
+ required: ["session_token", "heartbeat_interval"],
properties: {
- "version": { type: "string" },
- }*/
+ "session_token": { type: "string" },
+ "heartbeat_interval": { type: "number" },
+ },
},
},
},
- closeSchema,
+ {
+ properties: {
+ "type": { const: "continue" },
+ "data": {
+ additionalProperties: false,
+ required: ["last_seq", "heartbeat_interval"],
+ properties: {
+ "last_seq": { type: ["integer", "null"] },
+ "heartbeat_interval": { type: "number" },
+ },
+ },
+ },
+ },
+ {
+ properties: {
+ "type": { const: "invalidate" },
+ "data": { const: {} },
+ },
+ },
],
});
@@ -130,7 +136,19 @@ const clientHandshake = ajv.compile({
},
},
},
- closeSchema,
+ {
+ properties: {
+ "type": { const: "resume" },
+ "data": {
+ additionalProperties: false,
+ required: ["session_token", "last_seq"],
+ properties: {
+ "session_token": { type: "string" },
+ "last_seq": { type: ["integer", "null"] },
+ },
+ },
+ },
+ },
],
});
@@ -139,8 +157,8 @@ module.exports = {
compile: ajv.compile.bind(ajv),
message,
- close,
+ heartbeat,
serverHandshake,
clientHandshake,
-}
+};
diff --git a/master.js b/master.js
index 29e8b8cc..ef08616b 100644
--- a/master.js
+++ b/master.js
@@ -222,11 +222,15 @@ async function shutdown() {
}
for (let controlConnection of controlConnections) {
- controlConnection.connector.close("shutdown");
+ controlConnection.connector.close(1001, "Server Quit");
}
for (let slaveConnection of slaveConnections.values()) {
- slaveConnection.connector.close("shutdown");
+ slaveConnection.connector.close(1001, "Server Quit");
+ }
+
+ for (let socket of pendingSockets) {
+ socket.close(1001, "Server Quit");
}
if (httpServer) {
@@ -537,6 +541,12 @@ class SlaveConnection extends BaseConnection {
}
}
}
+
+ async shutdown(code, reason) {
+ await link.messages.shutdownConnection.send(this);
+ this.connector.close(code, reason);
+ await events.once(this.connector, "close");
+ }
}
const wss = new WebSocket.Server({
@@ -559,110 +569,277 @@ function isInteger(value) {
/**
* Connector for master server connections
*/
-class WebSocketServerConnector extends events.EventEmitter {
- constructor(socket) {
+class WebSocketServerConnector extends link.WebSocketBaseConnector {
+ constructor(socket, sessionId) {
super();
- this._seq = 1;
this._socket = socket;
+ this._sessionId = sessionId;
- this._socket.on('message', message => {
- this.emit('message', JSON.parse(message));
- });
- this._socket.on('close', (code, reason) => {
- this.emit('disconnect');
- });
+ // The following states are used in the server connector
+ // handshake: Wating for client to (re)connect.
+ // connected: Connection is online
+ // closing: Connection is in the process of being closed.
+ // closed: Connection has been closed
+ this._state = "handshake";
+ }
+
+ _check(expectedStates) {
+ if (!expectedStates.includes(this._state)) {
+ throw new Error(
+ `Expected state ${expectedStates} but state is ${this._state}`
+ );
+ }
}
/**
- * Send a message over the socket
+ * Send ready over the socket
*
- * @returns the sequence number of the message sent
+ * Sends the ready message over the socket to initiate the session.
*/
- send(type, data = {}) {
- this._socket.send(JSON.stringify({ seq: this._seq, type, data }));
- return this._seq++;
+ ready(sessionToken) {
+ this._heartbeatInterval = masterConfig.get("master.heartbeat_interval")
+ this._socket.send(JSON.stringify({
+ seq: null,
+ type: "ready",
+ data: {
+ session_token: sessionToken,
+ heartbeat_interval: this._heartbeatInterval,
+ },
+ }));
+
+ this._state = "connected";
+ this._attachSocketHandlers();
+ }
+
+ /**
+ * Continue connection with the given socket
+ *
+ * Terminates the current socket and contiunes the session over the
+ * socket given from the message sequence given.
+ */
+ continue(socket, lastSeq) {
+ this._socket.terminate();
+ this._socket = socket;
+
+ this._heartbeatInterval = masterConfig.get("master.heartbeat_interval")
+ this._socket.send(JSON.stringify({
+ seq: null,
+ type: "continue",
+ data: {
+ last_seq: this._lastReceivedSeq,
+ heartbeat_interval: this._heartbeatInterval,
+ },
+ }));
+
+ this._state = "connected";
+ this._attachSocketHandlers();
+ this._dropSendBufferSeq(lastSeq);
+ for (let message of this._sendBuffer) {
+ this._socket.send(JSON.stringify(message));
+ }
+ }
+
+ _attachSocketHandlers() {
+ this.startHeartbeat();
+
+ this._socket.on("close", (code, reason) => {
+ console.log(`SOCKET | Close (code: ${code}, reason: ${reason})`);
+ if (this._state === "closing") {
+ this._lastReceivedSeq = null;
+ this._sendBuffer.length = 0;
+ this._state = "closed";
+ activeConnections.delete(this._sessionId);
+
+ if (this._connected) {
+ this.emit("close");
+ this._connected = false;
+ }
+
+ } else {
+ this._state = "handshake";
+
+ if (this._connected) {
+ this.emit("dropped");
+ this._connected = false;
+ }
+ }
+
+ this.stopHeartbeat();
+ });
+
+ this._socket.on("error", err => {
+ // It's assumed that close is always called by ws
+ console.error("SOCKET | Error:", err);
+ });
+
+ this._socket.on("open", () => {
+ console.log("SOCKET | Open");
+ });
+ this._socket.on("ping", data => {
+ console.log(`SOCKET | Ping (data: ${data}`);
+ });
+ this._socket.on("pong", data => {
+ console.log(`SOCKET | Pong (data: ${data}`);
+ });
+
+ // Handle messages
+ this._socket.on("message", data => {
+ let message = JSON.parse(data);
+ if (["connected", "closing"].includes(this._state)) {
+ if (message.seq !== null) {
+ this._lastReceivedSeq = message.seq;
+ }
+
+ if (message.type === "heartbeat") {
+ this._processHeartbeat(message);
+
+ } else {
+ this.emit("message", message);
+ }
+
+ } else {
+ throw new Error(`Received message in unexpected state ${this._state}`);
+ }
+ });
}
/**
* Close the connection with the given reason.
*
- * Sends a close message and disconnects the connector.
+ * Sends a close frame and disconnects the connector.
*/
- close(reason) {
- this.send('close', { reason });
- this._socket.close();
- }
+ close(code, reason) {
+ if (this._state === "new") {
+ return;
+ }
- /**
- * Immediatly close the connection
- */
- disconnect() {
- this._socket.terminate();
+ this.stopHeartbeat();
+
+ if (["handshake", "connected"].includes(this._state)) {
+ this._socket.close(code, reason);
+ }
+
+ this._state = "closing";
}
}
-wss.on('connection', function (socket, req) {
- console.log(`SOCKET | new connection from ${socket.remoteAddress}`);
+let pendingSockets = new Set();
+let activeConnectors = new Map();
- // Start connection handshake
- let connector = new WebSocketServerConnector(socket);
- connector.send('hello', { version });
+wss.on("connection", function (socket, req) {
+ console.log(`SOCKET | new connection from ${req.socket.remoteAddress}`);
- // Handle socket handshake
- socket.once('message', (payload) => {
- payload = JSON.parse(payload);
+ // Start connection handshake.
+ socket.send(JSON.stringify({ seq: null, type: "hello", data: { version }}));
- // XXX prometheusWsUsageCounter.labels('message', "other").inc();
- if (!schema.clientHandshake(payload)) {
- console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid handshake`, payload);
- connector.close("Invalid handshake");
- return;
- }
+ function attachHandler() {
+ pendingSockets.add(socket);
- let { seq, type, data } = payload;
- if (["register_slave", "register_control"].includes(type)) {
- authenticate.check(data.token).then((result) => {
- if (result.ok) {
- if (type === "register_slave") {
- let connection = slaveConnections.get(data.id);
- if (connection) {
- console.log(`SOCKET | disconecting existing connection for slave ${data.id}`);
- connection.connector.close("Registered from another connection");
- }
+ let timeoutId = setTimeout(() => {
+ console.log(`SOCKET | closing ${req.socket.remoteAddress} after timing out on handshake`);
+ socket.close(1008, "Handshake timeout");
+ pendingSockets.delete(socket);
+ }, 30*1000);
- console.log(`SOCKET | registered slave ${data.id} version ${data.version}`);
- slaveConnections.set(data.id, new SlaveConnection(data, connector));
- connector.send('ready');
-
- } else { // type === "register_control"
- console.log(`SOCKET | registered control from ${req.socket.remoteAddress}`);
- controlConnections.push(new ControlConnection(data, connector));
- connector.send('ready');
- }
-
- } else {
- console.error(`SOCKET | authentication failed for ${req.socket.remoteAddress}`);
- connector.close("Authentication failed");
- }
- }).catch((err) => {
+ socket.once("message", (message) => {
+ clearTimeout(timeoutId);
+ pendingSockets.delete(socket);
+ handleHandshake(
+ message, socket, req, attachHandler
+ ).catch(err => {
console.error(`
-+---------------------------------------------------------------------+
-| Unexpected error occured while authenticating client, please report |
-| it to https://github.com/clusterio/factorioClusterio/issues |
-+---------------------------------------------------------------------+`
++----------------------------------------------------------------+
+| Unexpected error occured in WebSocket handshake, please report |
+| it to https://github.com/clusterio/factorioClusterio/issues |
++----------------------------------------------------------------+`
);
console.error(err);
- connection.close("Unexpected error");
+ socket.close(1011, "Unexpected error");
});
+ });
+ }
- } else if (type === "close") {
- console.log(`SOCKET | received close from ${req.socket.remoteAddress}: ${data.reason}`);
- connector.disconnect();
+ attachHandler();
+});
+
+let nextSessionId = 1;
+
+// Unique string for the session token audience
+let masterSession = `session-${Date.now()}`;
+
+// Handle socket handshake
+async function handleHandshake(message, socket, req, attachHandler) {
+ try {
+ message = JSON.parse(message);
+ } catch (err) {
+ console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid JSON`);
+ socket.close(1001, "Invalid JSON");
+ return;
+ }
+
+ if (!schema.clientHandshake(message)) {
+ console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid handshake:`, message);
+ socket.close(1001, "Bad handshake");
+ return;
+ }
+
+ let { seq, type, data } = message;
+
+ if (type === "resume") {
+ let connector;
+ try {
+ let payload = jwt.verify(
+ data.session_token,
+ masterConfig.get("master.auth_secret"),
+ { audience: masterSession }
+ );
+
+ connector = activeConnectors.get(payload.sid);
+ if (!connector) {
+ throw new Error();
+ }
+
+ } catch(err) {
+ socket.send(JSON.stringify({ seq: null, type: "invalidate", data: {}}));
+ attachHandler();
return;
}
- });
-});
+
+ connector.continue(socket, data.last_seq);
+ return
+ }
+
+ let result = await authenticate.check(data.token);
+
+ if (!result.ok) {
+ console.error(`SOCKET | authentication failed for ${req.socket.remoteAddress}`);
+ socket.close(4003, "Authentication failed");
+ return;
+ }
+
+ let sessionId = nextSessionId++;
+ let sessionToken = jwt.sign({ aud: masterSession, sid: sessionId, }, masterConfig.get("master.auth_secret"));
+ let connector = new WebSocketServerConnector(socket, sessionId);
+ activeConnectors.set(sessionId, connector);
+
+ if (type === "register_slave") {
+ let connection = slaveConnections.get(data.id);
+ if (connection) {
+ console.log(`SOCKET | disconecting existing connection for slave ${data.id}`);
+ await connection.shutdown();
+ }
+
+ console.log(`SOCKET | registered slave ${data.id} version ${data.version}`);
+ slaveConnections.set(data.id, new SlaveConnection(data, connector));
+
+ } else if (type === "register_control") {
+ console.log(`SOCKET | registered control from ${req.socket.remoteAddress}`);
+ controlConnections.push(new ControlConnection(data, connector));
+ }
+
+ connector.ready(sessionToken);
+}
// handle plugins on the master
var masterPlugins = new Map();
diff --git a/test/integration/index.js b/test/integration/index.js
index 81de9837..0dd11bae 100644
--- a/test/integration/index.js
+++ b/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,
diff --git a/test/integration/link.js b/test/integration/link.js
new file mode 100644
index 00000000..f1b1dba5
--- /dev/null
+++ b/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");
+ });
+});
diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js
index ea84094a..b800a679 100644
--- a/test/lib/link/connector.js
+++ b/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 }]);
diff --git a/test/lib/link/link.js b/test/lib/link/link.js
index 81dae963..2091305f 100644
--- a/test/lib/link/link.js
+++ b/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');
diff --git a/test/master.js b/test/master.js
index f37b143b..0c985e0a 100644
--- a/test/master.js
+++ b/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");
- });
- });
- });
});
diff --git a/test/mock.js b/test/mock.js
index f484f309..1fcefdab 100644
--- a/test/mock.js
+++ b/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 {