From d4e072f37605cd0f47c9314ebe0e17cf1f753c03 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 2 Apr 2020 21:34:41 +0200 Subject: [PATCH] Rewrite reconnection logic Implement proper session tracking and resume on reconnect for the WebSocket server. --- CHANGELOG.md | 2 + client.js | 26 +-- clusterctl.js | 42 +++- docs/assets/connection_flow.dot | 20 ++ docs/assets/connection_flow.svg | 99 +++++++++ docs/assets/generate.sh | 4 + docs/assets/reconnection_flow.dot | 23 +++ docs/assets/reconnection_flow.svg | 112 ++++++++++ docs/assets/server_flow.dot | 21 ++ docs/assets/server_flow.svg | 118 +++++++++++ docs/socket.md | 264 +++++++++++++++++++----- lib/config/definitions.js | 14 ++ lib/errors.js | 4 + lib/link/connectors.js | 285 +++++++++++++++++++++----- lib/link/link.js | 48 ++++- lib/link/messages.js | 12 ++ lib/schema.js | 98 +++++---- master.js | 327 +++++++++++++++++++++++------- test/integration/index.js | 8 +- test/integration/link.js | 70 +++++++ test/lib/link/connector.js | 87 +++++--- test/lib/link/link.js | 19 -- test/master.js | 10 - test/mock.js | 9 - 24 files changed, 1403 insertions(+), 319 deletions(-) create mode 100644 docs/assets/connection_flow.dot create mode 100644 docs/assets/connection_flow.svg create mode 100755 docs/assets/generate.sh create mode 100644 docs/assets/reconnection_flow.dot create mode 100644 docs/assets/reconnection_flow.svg create mode 100644 docs/assets/server_flow.dot create mode 100644 docs/assets/server_flow.svg create mode 100644 test/integration/link.js 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 @@ + + + + + + +Client Connection Flow + + + +start + +Start + + + +wait_hello + +Wait for hello + + + +start->wait_hello + + +Connect to /api/socket + + + +new_session + +New session + + + +wait_hello->new_session + + +Receive hello + + + +wait_ready + +Wait for ready + + + +new_session->wait_ready + + +Send register_* + + + +connected + +Connected + + + +wait_ready->connected + + +Receive ready + + + +abort + +Abort + + + +wait_ready->abort + + +Connection closed with 4003 + + + +lost_connection + +Connection lost or +unexpectedly closed +while connecting + + + +lost_connection->start + + +Wait random delay + + + 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 @@ + + + + + + +Client Reconnection Flow + + + +reconnect + +Reconnect + + + +wait_hello + +Wait for hello + + + +reconnect->wait_hello + + +Connect to /api/socket + + + +resume_session + +Resume session + + + +wait_hello->resume_session + + +Receive hello + + + +wait_continue + +Wait for continue + + + +resume_session->wait_continue + + +Send resume + + + +new_session + +New session + + + +wait_continue->new_session + + +Receive invalidate + + + +connected + +Connected + + + +wait_continue->connected + + +Receive ready + + + +abort + +Abort + + + +wait_continue->abort + + +Connection closed with 4003 + + + +lost_connection + +Connection lost or +unexpectedly closed +while reconnecting + + + +lost_connection->reconnect + + +Wait random delay + + + 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 @@ + + + + + + +Server Connection Flow + + + +new_connection + +New connection + + + +wait_handshake + +Wait for handshake + + + +new_connection->wait_handshake + + +Send hello + + + +validate_register + +Validate register + + + +wait_handshake->validate_register + + +Receive register_* + + + +validate_resume + +Validate resume + + + +wait_handshake->validate_resume + + +Receive resume + + + +close + +Close connection + + + +wait_handshake->close + + +Bad handshake +Timeout + + + +connected + +Connected + + + +validate_register->connected + + +Valid register: +Send ready + + + +validate_register->close + + +Bad token +Bad register + + + +validate_resume->wait_handshake + + +Bad session: +send invalidate + + + +validate_resume->connected + + +Valid resume: +Send continue + + + +validate_resume->close + + +Bad token +Bad resume + + + 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. + +![Client connection flowchart](assets/connection_flow.svg) + +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). + +![Client reconnection flowchart](assets/reconnection_flow.svg) + +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. + +![Client connection flowchart](assets/server_flow.svg) + +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 {