diff --git a/docs/assets/WsClientConnector.svg b/docs/assets/WsClientConnector.svg
new file mode 100644
index 00000000..bff8e243
--- /dev/null
+++ b/docs/assets/WsClientConnector.svg
@@ -0,0 +1,3 @@
+
+
+
\ No newline at end of file
diff --git a/docs/assets/WsServerConnector.svg b/docs/assets/WsServerConnector.svg
new file mode 100644
index 00000000..87ae7439
--- /dev/null
+++ b/docs/assets/WsServerConnector.svg
@@ -0,0 +1,3 @@
+
+
+
\ No newline at end of file
diff --git a/docs/assets/connection_flow.dot b/docs/assets/connection_flow.dot
deleted file mode 100644
index a8d9643c..00000000
--- a/docs/assets/connection_flow.dot
+++ /dev/null
@@ -1,20 +0,0 @@
-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.drawio b/docs/assets/connection_flow.drawio
new file mode 100644
index 00000000..817353e9
--- /dev/null
+++ b/docs/assets/connection_flow.drawio
@@ -0,0 +1,321 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/docs/assets/connection_flow.svg b/docs/assets/connection_flow.svg
index 95414122..d13c8e16 100644
--- a/docs/assets/connection_flow.svg
+++ b/docs/assets/connection_flow.svg
@@ -1,99 +1,3 @@
-
-
-
-
-
+
+
+
\ No newline at end of file
diff --git a/docs/assets/connector_flow.drawio b/docs/assets/connector_flow.drawio
new file mode 100644
index 00000000..45c688e0
--- /dev/null
+++ b/docs/assets/connector_flow.drawio
@@ -0,0 +1,1171 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/docs/assets/connector_states.svg b/docs/assets/connector_states.svg
new file mode 100644
index 00000000..f8363c4d
--- /dev/null
+++ b/docs/assets/connector_states.svg
@@ -0,0 +1,3 @@
+
+
+
\ No newline at end of file
diff --git a/docs/assets/generate.sh b/docs/assets/generate.sh
deleted file mode 100755
index 6c90f7d5..00000000
--- a/docs/assets/generate.sh
+++ /dev/null
@@ -1,4 +0,0 @@
-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
deleted file mode 100644
index 19324276..00000000
--- a/docs/assets/reconnection_flow.dot
+++ /dev/null
@@ -1,23 +0,0 @@
-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
deleted file mode 100644
index 22c7a4b4..00000000
--- a/docs/assets/reconnection_flow.svg
+++ /dev/null
@@ -1,112 +0,0 @@
-
-
-
-
-
diff --git a/docs/assets/server_flow.dot b/docs/assets/server_flow.dot
deleted file mode 100644
index c16e6db0..00000000
--- a/docs/assets/server_flow.dot
+++ /dev/null
@@ -1,21 +0,0 @@
-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
deleted file mode 100644
index 3dedab60..00000000
--- a/docs/assets/server_flow.svg
+++ /dev/null
@@ -1,118 +0,0 @@
-
-
-
-
-
diff --git a/docs/devs/protocol.md b/docs/devs/protocol.md
new file mode 100644
index 00000000..39625378
--- /dev/null
+++ b/docs/devs/protocol.md
@@ -0,0 +1,268 @@
+# Protocol
+
+Clusterio communicates using a simple homebrew message passing protocol based on JSON over WebSocket with limited support for reliable transport of messages.
+The protocol is implemented in [packages/lib/link](/packages/lib/link) and uses the `ws` library to handle the WebSocket connections on the Node.js side and the native WebSocket API on the browser side.
+
+Two general kinds of messages are supported: events and requests.
+Events are messages sent by one party and received by the other and offer no notification for the sender on whether it was actually received.
+Requests are pairs of request-response messages where one side sends the requests and the other side replies with a response.
+Apart from the handshake used to establish connections the protocol is directionless, and either party can send or receive both requests and events.
+
+Events and requests are defined as instances of the `Request` and `Event` classes, and specify a unique type string, who can send and receive the message, optional forwarding of the message and the format of the data payload as a JSON schema.
+All messages sent and received are validated with JSON schemas to catch potential errors in sending messages and avoid having to write lots of validation code in message handlers.
+
+Messages are sent by calling the `.send()` method of the message instance and passing a `Link` instance to send it over and the data to send.
+For the master server there's a `Link` instance for each slave and control connection stored in the `WsServer` class.
+For slaves the `Slave` instance is a link to the master server, and each `Instance` instance is an in-memory link to the `Slave` instance and has a corresponding `InstanceConnection` link stored on the `Slave` instance.
+For ctl and web_ui there's a `Control` instance that acts as the link to the master server.
+
+Messages received invokes message handlers that were registered when the link was brought up.
+These handlers are named after the name of the property the message is stored under with `EventHandler` appended for events and `RequestHandler` appended for requests.
+The handlers are defined on the link class itself for core messages, and on the plugin class in the case of plugin defined messages.
+
+
+## Connection flow
+
+Clients connect via WebSocket to the `/api/socket` endpoint and upon establishing connection to the server the connection is in the handshake phase and the server will send a `hello` message to the client.
+If the client has no previous session with the server it sends a `register_slave` or `register_control` message depending on what kind of client it is to the server.
+If the register is valid then the server will respond with a `ready` message with a session token and the connection has left the handshake phase, otherwise the server will close the connection with a WebSocket close code indicating the kind of failure.
+
+Upon leaving the handshake phase both parties should start sending heartbeat messages with the sequence number of the last received message.
+Messages sent over the link should be stored in a send buffer and only cleared from the send buffer when a heartbeat with a `last_seq` equal or or higher is received or the session is invalidated.
+
+If the connection goes stale or is unexpectedly closed the client can reastablish the same session by connecting to the server again and sending a `resume` message after received the `hello` from the server.
+If the session resume is successful then the server will respond with a `continue` message and the connection is active again.
+The client should then re-send all messages it has in its send buffer that is greater than the `last_seq` number in the `continue` message.
+Otherwise the server will respond with `invalidate` and the client is expected to drop its send buffer start a new session with `register_slave` or `register_control`.
+
+
+
+
+## Connector flow
+
+To manage the complexity of connecting, reconnecting, resuming or invalidating dropped sessions the WebSocket connection for a `Link` class is managed by a connector.
+There's three kinds of connectors defined: `WebSocketClientConnector`, `WebSocketServerConnector` and `VirtualConnector`.
+The virtual connector is for in-memory connections and doesn't have any connection state logic.
+
+For the WebSocket connectors there's four states they can exist in:
+- `closed` - The connector has just been created or has been explicitly closed.
+- `connecting` (client only) - The connector is in the process of trying to connect to the server.
+- `connected` - The connector is successfully connected.
+- `resuming` - The connection dropped and the connector is in the process of attempting to resume it.
+
+Along with the following events emitted when transition between states:
+- `connect` From `connecting` (client) or `closed` (server) to `connected`
+- `drop` - From `connected` to `resuming`
+- `resume` - From `resuming` to `connected`
+- `invalidate` - From `resuming` to `connecting` or `closed`
+- `close` - From any state to `closed`
+
+
+
+
+### WsClientConnector
+
+The client connector will keep trying to establish and mantain a connection after `.connect()` is called on it, stopping only in the case of authentication failure or `.close()`/`.setClosing()` being called.
+Other than that, the logic is pretty straight forward.
+
+
+
+
+### WsServerConnector
+
+The server connector doesn't do the work of connecting or validating the handshake.
+Instead it's handed over a WebSocket connection from `WsServer` once there's only the step of sending the final message left to do.
+Because of this there's no `connecting` state, and once a session times out from the `resuming` state it goes directly to closed.
+
+
+
+
+## Wire format
+
+All communication in the protocol is done via WebSocket text frames containing a JSON payload with the following properties:
+
+- `seq: integer | null` - The sequence number of the message, can only be null for handshake and heartbeats.
+- `type: string` - The type of message
+- `data: object` - Message data
+
+Before a connection is established it's in the handshake phase and the `seq` property is `null`.
+The `type` property is used to distinguish which of handshake, heartbeat, events or requests this message is and determines the expected format of the `data` property.
+
+
+### Handshake Messages
+
+The following messages are the only valid messages that can be sent during the handshake phase.
+
+
+#### `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".
+- `plugins: Object<string, string>` -
+ Object mapping plugin names to plugin version for plugins that are loaded on the master server.
+
+#### `register_slave`
+
+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.
+- `plugins: Object<string, string>` -
+ Object mapping plugin names to plugin version for plugins that are available on the slave.
+
+#### `register_control`
+
+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".
+
+#### `resume`
+
+Handshake for resuming an existing connection session where the connection had dropped unexpectedly.
+
+- `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.
+- `session_timeout: integer` - Session timeout in seconds.
+- `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.
+- `session_timeout: integer` - Session timeout in seconds.
+- `heartbeat_interval: number` - Rate client should send heartbeats at.
+
+
+### Heartbeat message
+
+Once connected as indicated by the `ready` and `continue` messages both sides of the connection starts sending heartbeats at the inverval given in the `ready` and `continue` messages.
+The heartbeat should contain the `seq` of the last received message, and receiving a heartbeat with a given sequence number means that the other party of the link has received all messages up to that seq number.
+
+
+#### `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.
+
+### 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 message types ending in `*_event` are events.
+The data payload of an event consists solely of event specific properties.
+
+See [packages/lib/link/messages.js](/packages/lib/link/messages.js) for the recognized core events and their contents.
+Plugins may defined their own events as well.
+
+#### `*_event`
+
+- any properties defined by the event type.
+
+
+### 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` 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.
+
+See [packages/lib/link/messages.js](/packages/lib/link/messages.js) for the recognized core requests and their contents.
+Plugins may define their own requests as well.
+
+#### `*_request`
+
+- any properties defined by the request type.
+
+#### `*_response` success
+
+- `seq: integer` - The seq number of the request message.
+- any additional properties defined by the request type.
+
+#### `*_response` error
+
+- `seq: integer` - The seq number of the request message.
+- `error: string` - Human readable text message describing the error.
+
+
+### Graceful Disconnect
+
+To ensure no data is lost when closing the link it's important to send a prepare disconnect 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.
+
+#### `prepare_disconnect_request`
+
+Request graceful disconnect 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.
+
+#### `prepare_disconnect_response`
+
+Signals the preparation for disconnect is complete and that the connection can be closed.
+The party that initiatied the prepare disconnect request may still send other requests/events over the link to finish its part of the disconnection proceedure.
+
+If the error response variant is received it should be logged if possible and treated as a success.
+
+
+### 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",
+ "session_timeout": 60,
+ "heartbeat_interval": 15
+ }
+}
+```
+
+Slave and master can now send events and requests over the connection.
diff --git a/docs/socket.md b/docs/socket.md
deleted file mode 100644
index 144b4bac..00000000
--- a/docs/socket.md
+++ /dev/null
@@ -1,225 +0,0 @@
-# WebSocket protocol
-
-This document describes the protocol used between the master and slaves, and unless you're developing Clusterio it's probably not going 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 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 either `register_*` or a `resume` message to set up the connection with the master.
-The exact flow is described below.
-
-
-### 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".
-- plugins - Object<string, string> -
- Object mapping plugin names to plugin version for plugins that are loaded on the master server.
-
-### `register_slave`
-
-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.
-- plugins - Object<string, string> -
- Object mapping plugin names to plugin version for plugins that are available on the slave.
-
-### `register_control`
-
-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".
-
-### `resume`
-
-Handshake for resuming an existing connection session where the connection had dropped unexpectedly.
-
-- 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
- }
-}
-```
-
-
-## 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 message types ending in `*_event` are events.
-The data payload of an event consists solely of event specific properties.
-
-See [packages/lib/link/messages.js](/packages/lib/link/messages.js) for the recognized events and their contents.
-
-
-## 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` 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.
-
-See [packages/lib/link/messages.js](/packages/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.
-
-### `*_response` error
-
-- seq - integer - The seq number of the request message.
-- error - string - Human readable text message describing the error.
-
-
-## Graceful Disconnect
-
-To ensure no data is lost when closing the link it's important to send a prepare disconnect 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.
-
-### `prepare_disconnect_request`
-
-Request graceful disconnect 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.
-
-### `prepare_disconnect_response`
-
-Signals the preparation for disconnect is complete and that the connection can be closed.
-The party that initiatied the prepare disconnect request may still send other requests/events over the link to finish its part of the disconnection proceedure.
-
-If the error response variant is received it should be logged if possible and treated as a success.
diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js
index bc7b7db3..39fe0122 100755
--- a/packages/ctl/ctl.js
+++ b/packages/ctl/ctl.js
@@ -866,13 +866,13 @@ debugCommands.add(new libCommand.Command({
* @private
*/
class ControlConnector extends libLink.WebSocketClientConnector {
- constructor(url, reconnectDelay, tlsCa, token) {
- super(url, reconnectDelay, tlsCa);
+ constructor(url, maxReconnectDelay, tlsCa, token) {
+ super(url, maxReconnectDelay, tlsCa);
this._token = token;
}
register() {
- logger.verbose("SOCKET | registering control");
+ logger.verbose("Connector | registering control");
this.sendHandshake("register_control", {
token: this._token,
agent: "clusterioctl",
@@ -946,8 +946,6 @@ class Control extends libLink.Link {
}
async shutdown() {
- this.connector.setTimeout(30);
-
try {
await libLink.messages.prepareDisconnect.send(this);
} catch (err) {
@@ -956,7 +954,7 @@ class Control extends libLink.Link {
}
}
- await this.connector.close(1001, "Control Quit");
+ await this.connector.close(1000, "Control Quit");
}
}
@@ -1120,7 +1118,7 @@ async function startControl() {
let controlConnector = new ControlConnector(
controlConfig.get("control.master_url"),
- controlConfig.get("control.reconnect_delay"),
+ controlConfig.get("control.max_reconnect_delay"),
tlsCa,
controlConfig.get("control.master_token")
);
diff --git a/packages/lib/config/definitions.js b/packages/lib/config/definitions.js
index 043611f9..0df15e2f 100644
--- a/packages/lib/config/definitions.js
+++ b/packages/lib/config/definitions.js
@@ -85,14 +85,14 @@ MasterGroup.define({
title: "Heartbeat Interval",
description: "Interval heartbeats are sent out on WebSocket connections",
type: "number",
- initial_value: 45,
+ initial_value: 15,
});
MasterGroup.define({
- name: "connector_shutdown_timeout",
- title: "Connector Shutdown Timeout",
- description: "Timeout in seconds for connectors to properly disconnect on shutdown",
+ name: "session_timeout",
+ title: "Session Timeout",
+ description: "Time in seconds before giving up resuming a dropped WebSocket session.",
type: "number",
- initial_value: 30,
+ initial_value: 60,
});
MasterGroup.define({
name: "metrics_timeout",
@@ -184,11 +184,11 @@ SlaveGroup.define({
initial_value: "localhost",
});
SlaveGroup.define({
- name: "reconnect_delay",
- title: "Reconnect Delay",
+ name: "max_reconnect_delay",
+ title: "Max Reconnect Delay",
description: "Maximum delay to wait before attempting to reconnect WebSocket",
type: "number",
- initial_value: 5,
+ initial_value: 60,
});
SlaveGroup.finalize();
@@ -358,11 +358,11 @@ ControlGroup.define({
optional: true,
});
ControlGroup.define({
- name: "reconnect_delay",
- title: "Reconnect Delay",
+ name: "max_reconnect_delay",
+ title: "Max Reconnect Delay",
description: "Maximum delay to wait before attempting to reconnect WebSocket",
type: "number",
- initial_value: 2,
+ initial_value: 60,
});
ControlGroup.finalize();
diff --git a/packages/lib/link/connectors.js b/packages/lib/link/connectors.js
index d7a4fd2f..3ae4235b 100644
--- a/packages/lib/link/connectors.js
+++ b/packages/lib/link/connectors.js
@@ -6,6 +6,7 @@ const WebSocket = require("isomorphic-ws");
const libSchema = require("../schema");
const libErrors = require("../errors");
const { logger } = require("../logging");
+const ExponentialBackoff = require("../ExponentialBackoff");
/**
@@ -15,9 +16,14 @@ const { logger } = require("../logging");
* @memberof module:lib/link
*/
class WebSocketBaseConnector extends events.EventEmitter {
- constructor() {
+ constructor(sessionTimeout) {
super();
+ this._sessionTimeout = sessionTimeout;
+
+ // One of closed, connecting (client only), connected and resuming.
+ this._state = "closed";
+ this._closing = false;
this._socket = null;
this._seq = 1;
this._lastHeartbeat = null;
@@ -27,6 +33,26 @@ class WebSocketBaseConnector extends events.EventEmitter {
this._sendBuffer = [];
}
+ _reset() {
+ this._state = "closed";
+ this._closing = false;
+ this._socket = null;
+ this._seq = 1;
+ this._lastHeartbeat = null;
+ this._heartbeatId = null;
+ this._heartbeatInterval = null;
+ this._lastReceivedSeq = null;
+ this._sendBuffer.length = 0;
+ }
+
+ _check(...expectedStates) {
+ if (!expectedStates.includes(this._state)) {
+ throw new Error(
+ `Expected state ${expectedStates} but state is ${this._state}`
+ );
+ }
+ }
+
_dropSendBufferSeq(seq) {
if (seq === null) {
return;
@@ -44,12 +70,10 @@ class WebSocketBaseConnector extends events.EventEmitter {
}
_doHeartbeat() {
- this._check(["connected", "closing"]);
+ this._check("connected");
if (Date.now() - this._lastHeartbeat > 2000 * this._heartbeatInterval) {
- logger.verbose("SOCKET | closing after heartbeat timed out");
+ logger.verbose("Connector | closing after heartbeat timed out");
this._socket.close(1008, "Heartbeat timeout");
- clearInterval(this._heartbeatId);
- this._heartbeatId = null;
} else {
this._socket.send(JSON.stringify({
@@ -62,7 +86,7 @@ class WebSocketBaseConnector extends events.EventEmitter {
_processHeartbeat(message) {
if (!libSchema.heartbeat(message)) {
- logger.warn("SOCKET | closing after received invalid heartbeat");
+ logger.warn("Connector | closing after received invalid heartbeat");
this._socket.close(1002, "Invalid heartbeat");
return;
}
@@ -91,6 +115,7 @@ class WebSocketBaseConnector extends events.EventEmitter {
if (this._heartbeatId) {
clearInterval(this._heartbeatId);
this._heartbeatId = null;
+ this._lastHeartbeat = null;
}
}
@@ -102,14 +127,14 @@ class WebSocketBaseConnector extends events.EventEmitter {
* @returns {number} the sequence number of the message sent
*/
send(type, data) {
- if (!["handshake", "connected", "closing"].includes(this._state)) {
+ if (!["connected", "resuming"].includes(this._state)) {
throw new libErrors.SessionLost("No session");
}
let seq = this._seq;
this._seq += 1;
let message = { seq, type, data };
this._sendBuffer.push(message);
- if (["connected", "closing"].includes(this._state)) {
+ if (this._state === "connected") {
this._socket.send(JSON.stringify(message));
}
return seq;
@@ -118,11 +143,11 @@ class WebSocketBaseConnector extends events.EventEmitter {
/**
* Set connection state to closing
*
- * Signal the connection is about tho close and that once the close
+ * Signal the connection is about to close and that once the close
* frame is received it should be considered finished.
*/
setClosing() {
- this._state = "closing";
+ throw new Error("Abstract function");
}
/**
@@ -136,7 +161,7 @@ class WebSocketBaseConnector extends events.EventEmitter {
* True if the connection is closing down or closed
*/
get closing() {
- return ["closing", "closed"].includes(this._state);
+ return this._closing || this._state === "closed";
}
}
@@ -147,41 +172,32 @@ class WebSocketBaseConnector extends events.EventEmitter {
* @memberof module:lib/link
*/
class WebSocketClientConnector extends WebSocketBaseConnector {
- constructor(url, reconnectDelay, tlsCa = null) {
- super();
+ constructor(url, maxReconnectDelay, tlsCa = null) {
+ super(null);
this._url = url;
+ this._backoff = new ExponentialBackoff({ max: maxReconnectDelay });
this._tlsCa = tlsCa;
// The following states are used in the client connector
- // new: Not connected
- // handshake: Attempting to (re)connect to server.
+ // closed: Not connected
+ // connecting: Attempting to connect to server.
+ // resuming: Attempting to resume an existing connection to the 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;
- this._timeout = 15 * 60;
- this._startedReconnect = null;
+ this._startedResuming = null;
}
- _check(expectedStates) {
- if (!expectedStates.includes(this._state)) {
- throw new Error(
- `Expected state ${expectedStates} but state is ${this._state}`
- );
- }
- }
-
- setTimeout(timeout) {
- this._timeout = timeout;
-
- if (this._reconnectId && this._timeout < this._reconnectDelay / 2) {
- clearTimeout(this._reconnectId);
- this.reconnect();
- }
+ _reset() {
+ clearTimeout(this._reconnectId);
+ this._reconnectId = null;
+ this._lastReceivedSeq = null;
+ this._sessionToken = null;
+ this._sessionTimeout = null;
+ this._sendBuffer.length = 0;
+ this._startedResuming = null;
+ super._reset();
}
/**
@@ -194,7 +210,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
* @param {Object} data - Data to send with message.
*/
sendHandshake(type, data) {
- this._check(["handshake"]);
+ this._check("connecting", "resuming");
this._socket.send(JSON.stringify({ seq: null, type, data }));
}
@@ -207,34 +223,53 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
* @param {string} reason - WebSocket close reason.
*/
async close(code, reason) {
- if (this._state === "new") {
+ if (this._state === "closed") {
return;
}
- if (this._reconnectId) {
- clearTimeout(this._reconnectId);
- this._reconnectId = null;
- this._lastReceivedSeq = null;
- this._sessionToken = null;
- this._sendBuffer.length = 0;
- this._state = "new";
- this._connected = false;
+ this._closing = true;
+ if (this._state === "connected" || this._socket) {
+ this._socket.close(code, reason);
+ await events.once(this, "close");
+
+ } else {
+ this._reset();
this.emit("close");
+ }
+ }
+
+ /**
+ * Notify the connection is expected to close
+ *
+ * Sets the internal flag to signal not to attempt to resume the
+ * connection if the server closes it. This will close the connection
+ * if it's currently attempting to connect or resume.
+ */
+ setClosing() {
+ if (this._state === "closed") {
return;
}
- this.stopHeartbeat();
- this._state = "closing";
- this._socket.close(code, reason);
- await events.once(this, "close");
+ this._closing = true;
+ if (this._state === "connected") {
+ return;
+ }
+
+ if (this._socket) {
+ this._socket.close(1000, "Connector closing");
+
+ } else {
+ this._reset();
+ this.emit("close");
+ }
}
/**
* Connect to the master server
*/
async connect() {
- this._check(["new"]);
- this._state = "handshake";
+ this._check("closed");
+ this._state = "connecting";
this._doConnect();
@@ -242,12 +277,12 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
await events.once(this, "connect");
}
- async _doConnect() {
+ _doConnect() {
let url = new URL(this._url);
url.pathname += "api/socket";
// Open WebSocket to master
- logger.verbose(`SOCKET | connecting to ${url}`);
+ logger.verbose(`Connector | connecting to ${url}`);
// eslint-disable-next-line node/no-process-env
if (process.env.APP_ENV === "browser") {
@@ -266,69 +301,89 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
* Attempt re-establishing the connecting again
*/
reconnect() {
+ if (this._socket) {
+ throw new Error("Cannot reconnect while socket is open");
+ }
+
if (this._reconnectId !== null) {
- return;
+ logger.error("Unexpected double call to reconnect");
}
- if (this._startedReconnect === null) {
- this._startedReconnect = Date.now();
+ let delay = this._backoff.delay();
+ if (
+ this._state === "resuming"
+ && this._startedResuming + this._sessionTimeout * 1000 < Date.now() + delay
+ ) {
+ logger.error("Connector | Session timed out trying to resume");
+ this._reset();
+ this._state = "connecting";
+ this.emit("invalidate");
}
-
- if (this._startedReconnect + this._timeout * 1000 < Date.now()) {
- logger.error("SOCKET | Timed out trying to reconnect");
- this._heartbeatInterval = null;
- this._lastReceivedSeq = null;
- this._sessionToken = null;
- this._sendBuffer.length = 0;
- this._state = "new";
- this._connected = false;
- this.emit("close");
- return;
- }
-
- let delay = Math.random() * this._reconnectDelay;
logger.verbose(
- `SOCKET | waiting ${delay.toLocaleString("en", { maximumFractionDigits: 2 })} seconds for reconnect`
+ `Connector | waiting ${(Math.round(delay / 10) / 100)} seconds for reconnect`
);
this._reconnectId = setTimeout(() => {
this._reconnectId = null;
this._doConnect();
- }, delay * 1000);
+ }, delay);
+ this._reconnectTime = Date.now() + delay;
}
_attachSocketHandlers() {
this._socket.onclose = event => {
- logger.verbose(`SOCKET | Close (code: ${event.code}, reason: ${event.reason})`);
+ const previousState = this._state;
+
// Authentication failed
if (event.code === 4003) {
this.emit("error", new libErrors.AuthenticationFailed(event.reason));
- this._state = "closing";
+ this._closing = true;
}
- if (this._state === "closing") {
- this._lastReceivedSeq = null;
- this._sessionToken = null;
- this._sendBuffer.length = 0;
- this._state = "new";
- this._connected = false;
- this.emit("close");
+ this._socket = null;
+ if (this._state === "connected") {
+ this.stopHeartbeat();
+ if (this._closing) {
+ this._reset();
+ this.emit("close");
+
+ } else {
+ this._state = "resuming";
+ this._startedResuming = Date.now();
+ this.reconnect();
+ this.emit("drop");
+ }
} else {
- this._state = "handshake";
- this.reconnect();
- if (this._connected) {
- this._connected = false;
- this.emit("drop");
+ // eslint-disable-next-line no-lonely-if
+ if (this._closing) {
+ this._reset();
+ this.emit("close");
+
+ } else {
+ this.reconnect();
}
}
- this.stopHeartbeat();
+ // Log must be sent after state change is complete
+ let message = `Connector | Close (code: ${event.code}, reason: ${event.reason})`;
+ if (previousState === "connected" && event.code !== 1000) {
+ logger.info(message);
+ } else {
+ logger.verbose(message);
+ }
};
this._socket.onerror = event => {
// It's assumed that close is always called by ws
let code = !event.error ? "" : `, code: ${event.error.code}`;
- logger.error(`SOCKET | Error: ${event.message || "unknown error"}${code}`);
+ let message = `Connector | Socket error: ${event.message || "unknown error"}${code}`;
+ if (this._state === "connected") {
+ logger.error(message);
+ } else {
+ // Don't log as error if not actually connected, prevents
+ // log spam during reconnecting attempts.
+ logger.verbose(message);
+ }
if (event.error) {
// Abort connection if certificate validation failed
if ([
@@ -337,22 +392,22 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
"ERR_TLS_CERT_ALTNAME_INVALID",
].includes(event.error.code)) {
this.emit("error", new libErrors.AuthenticationFailed(event.error.message));
- this._state = "closing";
+ this._closing = true;
}
}
};
this._socket.onopen = () => {
- logger.verbose("SOCKET | Open");
+ logger.verbose("Connector | Open");
};
// Handle messages
this._socket.onmessage = event => {
let message = JSON.parse(event.data);
- if (this._state === "handshake") {
+ if (["connecting", "resuming"].includes(this._state)) {
this._processHandshake(message);
- } else if (["connected", "closing"].includes(this._state)) {
+ } else if (this._state === "connected") {
if (message.seq !== null) {
this._lastReceivedSeq = message.seq;
}
@@ -372,17 +427,17 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
_processHandshake(message) {
if (!libSchema.serverHandshake(message)) {
- logger.verbose("SOCKET | closing after received invalid handshake");
+ logger.warn("Connector | closing after received invalid handshake");
this._socket.close(1002, "Invalid handshake");
return;
}
let { seq, type, data } = message;
if (type === "hello") {
- logger.verbose(`SOCKET | received hello from master version ${data.version}`);
+ logger.verbose(`Connector | received hello from master version ${data.version}`);
this.emit("hello", data);
if (this._sessionToken) {
- logger.verbose("SOCKET | Attempting resume");
+ logger.verbose("Connector | Attempting resume");
this.sendHandshake("resume", {
session_token: this._sessionToken,
last_seq: this._lastReceivedSeq,
@@ -392,36 +447,39 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
}
} else if (type === "ready") {
- logger.verbose("SOCKET | received ready from master");
+ logger.verbose("Connector | received ready from master");
this._state = "connected";
this._sessionToken = data.session_token;
+ this._sessionTimeout = data.session_timeout;
this._heartbeatInterval = data.heartbeat_interval;
this.startHeartbeat();
for (let bufferedMessage of this._sendBuffer) {
this._socket.send(JSON.stringify(bufferedMessage));
}
- this._startedReconnect = null;
- this._connected = true;
this.emit("connect", data);
} else if (type === "continue") {
- logger.verbose("SOCKET | resuming existing session");
+ logger.info("Connector | resuming existing session");
this._state = "connected";
this._heartbeatInterval = data.heartbeat_interval;
+ this._sessionTimeout = data.session_timeout;
this.startHeartbeat();
this._dropSendBufferSeq(data.last_seq);
for (let bufferedMessage of this._sendBuffer) {
this._socket.send(JSON.stringify(bufferedMessage));
}
- this._startedReconnect = null;
- this._connected = true;
+ this._startedResuming = null;
this.emit("resume");
} else if (type === "invalidate") {
- this._heartbeatInterval = null;
+ logger.warn("Connector | session invalidated by master");
+ this._state = "connecting";
+ this._seq = 1;
this._lastReceivedSeq = null;
this._sessionToken = null;
+ this._sessionTimeout = null;
this._sendBuffer.length = 0;
+ this._startedResuming = null;
this.emit("invalidate");
this.register();
}
diff --git a/packages/lib/schema.js b/packages/lib/schema.js
index 7ceea00e..6753fd69 100644
--- a/packages/lib/schema.js
+++ b/packages/lib/schema.js
@@ -98,9 +98,10 @@ const serverHandshake = ajv.compile({
"type": { const: "ready" },
"data": {
additionalProperties: false,
- required: ["session_token", "heartbeat_interval"],
+ required: ["session_token", "session_timeout", "heartbeat_interval"],
properties: {
"session_token": { type: "string" },
+ "session_timeout": { type: "number" },
"heartbeat_interval": { type: "number" },
"account": {
additionalProperties: false,
@@ -118,9 +119,10 @@ const serverHandshake = ajv.compile({
"type": { const: "continue" },
"data": {
additionalProperties: false,
- required: ["last_seq", "heartbeat_interval"],
+ required: ["last_seq", "session_timeout", "heartbeat_interval"],
properties: {
"last_seq": { type: ["integer", "null"] },
+ "session_timeout": { type: "number" },
"heartbeat_interval": { type: "number" },
},
},
diff --git a/packages/master/src/WsServer.js b/packages/master/src/WsServer.js
index c0128968..3c9b0238 100644
--- a/packages/master/src/WsServer.js
+++ b/packages/master/src/WsServer.js
@@ -2,6 +2,7 @@
const jwt = require("jsonwebtoken");
const WebSocket = require("ws");
+const libErrors = require("@clusterio/lib/errors");
const { logger } = require("@clusterio/lib/logging");
const libPrometheus = require("@clusterio/lib/prometheus");
const libSchema = require("@clusterio/lib/schema");
@@ -64,12 +65,10 @@ class WsServer {
let disconnectTasks = [];
for (let controlConnection of this.controlConnections) {
- controlConnection.connector.setTimeout(this.master.config.get("master.connector_shutdown_timeout"));
disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit"));
}
for (let slaveConnection of this.slaveConnections.values()) {
- slaveConnection.connector.setTimeout(this.master.config.get("master.connector_shutdown_timeout"));
disconnectTasks.push(slaveConnection.disconnect(1001, "Server Quit"));
}
@@ -254,8 +253,9 @@ ${err.stack}`
{ aud: this.sessionAud, sid: sessionId },
Buffer.from(this.master.config.get("master.auth_secret"), "base64"),
);
+ let sessionTimeout = this.master.config.get("master.session_timeout");
let heartbeatInterval = this.master.config.get("master.heartbeat_interval");
- let connector = new WsServerConnector(socket, sessionId, heartbeatInterval);
+ let connector = new WsServerConnector(sessionId, sessionTimeout, heartbeatInterval);
this.activeConnectors.set(sessionId, connector);
connector.on("close", () => {
this.activeConnectors.delete(connector);
@@ -276,6 +276,8 @@ ${err.stack}`
if (this.slaveConnections.get(data.id) === connection) {
this.slaveConnections.delete(data.id);
this.master.slaveUpdated(this.master.slaves.get(data.id));
+ } else {
+ logger.warn("Unlisted SlaveConnection closed");
}
});
this.slaveConnections.set(data.id, connection);
@@ -286,9 +288,11 @@ ${err.stack}`
logger.verbose(`WsServer | registered control from ${req.socket.remoteAddress}`);
let connection = new ControlConnection(data, connector, this.master, user);
connector.on("close", () => {
- let index = this.controlConnections.indexOf(this);
+ let index = this.controlConnections.indexOf(connection);
if (index !== -1) {
this.controlConnections.splice(index, 1);
+ } else {
+ logger.warn("Unlisted ControlConnection closed");
}
});
this.controlConnections.push(connection);
@@ -297,7 +301,7 @@ ${err.stack}`
};
}
- connector.ready(sessionToken, additionalReadyData);
+ connector.ready(socket, sessionToken, additionalReadyData);
}
}
diff --git a/packages/master/src/WsServerConnector.js b/packages/master/src/WsServerConnector.js
index 721e2c89..bd58003d 100644
--- a/packages/master/src/WsServerConnector.js
+++ b/packages/master/src/WsServerConnector.js
@@ -12,28 +12,24 @@ const { logger } = require("@clusterio/lib/logging");
* @alias module:master/src/WsServerConnector
*/
class WsServerConnector extends libLink.WebSocketBaseConnector {
- constructor(socket, sessionId, heartbeatInterval) {
- super();
+ constructor(sessionId, sessionTimeout, heartbeatInterval) {
+ super(sessionTimeout);
- this._socket = socket;
+ this._socket = null;
this._sessionId = sessionId;
this._heartbeatInterval = heartbeatInterval;
+ this._timeoutId = null;
// The following states are used in the server connector
- // handshake: Waiting for client to (re)connect.
+ // closed: Connection is closed
// connected: Connection is online
- // closing: Connection is in the process of being closed.
- // closed: Connection has been closed
- this._state = "handshake";
- this._timeout = 15 * 60;
+ // resuming: Waiting for client to resume.
}
- _check(expectedStates) {
- if (!expectedStates.includes(this._state)) {
- throw new Error(
- `Expected state ${expectedStates} but state is ${this._state}`
- );
- }
+ _reset() {
+ clearTimeout(this._timeoutId);
+ this._timeoutId = null;
+ super._reset();
}
/**
@@ -41,17 +37,20 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
*
* Sends the ready message over the socket to initiate the session.
*
+ * @param {Object} socket - WebSocket connection to client.
* @param {string} sessionToken -
* the session token to send to the client.
* @param {Object=} additionalData -
* extra properties to send along the ready message.
*/
- ready(sessionToken, additionalData) {
+ ready(socket, sessionToken, additionalData) {
+ this._socket = socket;
this._socket.send(JSON.stringify({
seq: null,
type: "ready",
data: {
session_token: sessionToken,
+ session_timeout: this._sessionTimeout,
heartbeat_interval: this._heartbeatInterval,
...additionalData,
},
@@ -77,7 +76,7 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
// stale connection. If this is the case the connector will have to
// wait for the previous socket to close down before it can continue
// from a new socket, or there will be overlapping events from both.
- if (["connected", "closing"].includes(this._state)) {
+ if (this._state === "connected") {
this._socket.close(4003, "Session Hijacked");
// Correctly waiting for the connector to close and then
@@ -102,6 +101,7 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
type: "continue",
data: {
last_seq: this._lastReceivedSeq,
+ session_timeout: this._sessionTimeout,
heartbeat_interval: this._heartbeatInterval,
},
}));
@@ -115,63 +115,44 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
this.emit("resume");
}
- setTimeout(timeout) {
- if (this._timeoutId) {
- clearTimeout(this._timeoutId);
- }
-
- this._timeoutId = setTimeout(() => { this._timedOut(); }, timeout * 1000);
- this._timeout = timeout;
- }
-
_timedOut() {
- logger.verbose("SOCKET | Connection timed out");
- this._timeoutId = null;
- this._lastReceivedSeq = null;
- this._sendBuffer.length = 0;
- this._state = "closed";
+ logger.verbose("Connector | Connection timed out");
+ this._reset();
this.emit("close");
- this.emit("invalidate");
}
_attachSocketHandlers() {
this.startHeartbeat();
this._socket.on("close", (code, reason) => {
- logger.verbose(`SOCKET | Close (code: ${code}, reason: ${reason})`);
- if (this._state === "closing") {
- this._lastReceivedSeq = null;
- this._sendBuffer.length = 0;
- this._state = "closed";
+ logger.verbose(`Connector | Close (code: ${code}, reason: ${reason})`);
+ this.stopHeartbeat();
+
+ if (this._closing) {
+ this._reset();
this.emit("close");
- if (this._timeoutId) {
- clearTimeout(this._timeoutId);
- this._timeoutId = null;
- }
-
} else {
- this._state = "handshake";
+ this._state = "resuming";
this.emit("drop");
- this._timeoutId = setTimeout(() => { this._timedOut(); }, this._timeout * 1000);
+ this._timeoutId = setTimeout(() => { this._timedOut(); }, this._sessionTimeout * 1000);
}
- this.stopHeartbeat();
});
this._socket.on("error", err => {
// It's assumed that close is always called by ws
- logger.verbose("SOCKET | Error:", err);
+ logger.verbose("Connector | Error:", err);
});
this._socket.on("open", () => {
- logger.verbose("SOCKET | Open");
+ logger.verbose("Connector | Open");
});
this._socket.on("ping", data => {
- logger.verbose(`SOCKET | Ping (data: ${data}`);
+ logger.verbose(`Connector | Ping (data: ${data}`);
});
this._socket.on("pong", data => {
- logger.verbose(`SOCKET | Pong (data: ${data}`);
+ logger.verbose(`Connector | Pong (data: ${data}`);
});
// Handle messages
@@ -208,11 +189,34 @@ class WsServerConnector extends libLink.WebSocketBaseConnector {
return;
}
- this.stopHeartbeat();
- this._state = "closing";
+ if (this._state === "resuming") {
+ this._reset();
+ this.emit("close");
+ return;
+ }
+
+ this._closing = true;
this._socket.close(code, reason);
await events.once(this, "close");
}
+
+ /**
+ * Notify the connection is expected to close
+ *
+ * Sets the internal flag to signal not to wait for the client to
+ * reconnect and resume the connection if the client closes it. This
+ * will close the connection if it's currently waiting for the client to
+ * resume it.
+ */
+ setClosing() {
+ if (this._state === "resuming") {
+ this._reset();
+ this.emit("close");
+ return;
+ }
+
+ this._closing = true;
+ }
}
module.exports = WsServerConnector;
diff --git a/packages/slave/slave.js b/packages/slave/slave.js
index 948df556..ed9e9ac8 100755
--- a/packages/slave/slave.js
+++ b/packages/slave/slave.js
@@ -972,7 +972,7 @@ class SlaveConnector extends libLink.WebSocketClientConnector {
constructor(slaveConfig, tlsCa, pluginInfos) {
super(
slaveConfig.get("slave.master_url"),
- slaveConfig.get("slave.reconnect_delay"),
+ slaveConfig.get("slave.max_reconnect_delay"),
tlsCa
);
this.slaveConfig = slaveConfig;
@@ -980,7 +980,7 @@ class SlaveConnector extends libLink.WebSocketClientConnector {
}
register() {
- logger.info("SOCKET | registering slave");
+ logger.info("Connector | registering slave");
let plugins = {};
for (let pluginInfo of this.pluginInfos) {
plugins[pluginInfo.name] = pluginInfo.version;
@@ -1598,7 +1598,6 @@ class Slave extends libLink.Link {
return;
}
this._shuttingDown = true;
- this.connector.setTimeout(30);
try {
await libLink.messages.prepareDisconnect.send(this);
@@ -1612,7 +1611,7 @@ class Slave extends libLink.Link {
for (let instanceId of this.instanceConnections.keys()) {
await this.stopInstance(instanceId);
}
- await this.connector.close(1001, "Slave Shutdown");
+ await this.connector.close(1000, "Slave Shutdown");
// Clear silly interval in pidfile library.
pidusage.clear();
@@ -1862,6 +1861,7 @@ async function startSlave() {
});
await slaveConnector.connect();
+ logger.info("Started slave");
}
module.exports = {
diff --git a/packages/web_ui/src/bootstrap.jsx b/packages/web_ui/src/bootstrap.jsx
index dfef2d58..213715b4 100644
--- a/packages/web_ui/src/bootstrap.jsx
+++ b/packages/web_ui/src/bootstrap.jsx
@@ -85,8 +85,7 @@ export default async function bootstrap() {
let wsUrl = new URL(window.webRoot, document.location);
wsUrl.protocol = wsUrl.protocol.replace("http", "ws");
- let controlConnector = new ControlConnector(wsUrl, 10);
- controlConnector.setTimeout(15);
+ let controlConnector = new ControlConnector(wsUrl, 120);
let control = new Control(controlConnector, plugins);
ReactDOM.render(, document.getElementById("root"));
diff --git a/packages/web_ui/src/util/websocket.jsx b/packages/web_ui/src/util/websocket.jsx
index c0be3ef2..d7a1443f 100644
--- a/packages/web_ui/src/util/websocket.jsx
+++ b/packages/web_ui/src/util/websocket.jsx
@@ -8,8 +8,8 @@ import packageJson from "../../package.json";
* @private
*/
export class ControlConnector extends libLink.WebSocketClientConnector {
- constructor(url, reconnectDelay) {
- super(url, reconnectDelay);
+ constructor(url, maxReconnectDelay) {
+ super(url, maxReconnectDelay);
this.token = null;
}
@@ -18,7 +18,7 @@ export class ControlConnector extends libLink.WebSocketClientConnector {
throw new Error("Token not set");
}
- logger.verbose("SOCKET | registering control");
+ logger.verbose("Connector | registering control");
this.sendHandshake("register_control", {
token: this.token,
agent: "Clusterio Web UI",
diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js
index 812864eb..c5007d57 100644
--- a/test/integration/clusterio.js
+++ b/test/integration/clusterio.js
@@ -46,7 +46,7 @@ describe("Integration of Clusterio", function() {
connectorA.token = controlToken;
let controlA = new TestControl(connectorA);
await connectorA.connect();
- connectorA._state = "closing";
+ connectorA._closing = true;
connectorA.stopHeartbeat();
connectorA.on("error", () => {});
@@ -54,10 +54,12 @@ describe("Integration of Clusterio", function() {
connectorB.token = controlToken;
let controlB = new TestControl(connectorB);
connectorB._sessionToken = connectorA._sessionToken;
- connectorB._state = "handshake";
- await connectorB._doConnect();
+ connectorB._sessionTimeout = connectorA._sessionTimeout;
+ connectorB._startedResuming = Date.now();
+ connectorB._state = "resuming";
+ connectorB._doConnect();
await events.once(connectorB, "resume");
- await connectorB.close();
+ await connectorB.close(1000, "");
});
});
});
@@ -84,7 +86,7 @@ describe("Integration of Clusterio", function() {
let slaveProcess;
try {
slaveProcess = await spawn(
- "alt-slave:", `node ../../packages/slave run --config ${config}`, /SOCKET \| registering slave/
+ "alt-slave:", `node ../../packages/slave run --config ${config}`, /Started slave/
);
} finally {
if (slaveProcess) {
diff --git a/test/integration/index.js b/test/integration/index.js
index 6994ab7d..5dc04010 100644
--- a/test/integration/index.js
+++ b/test/integration/index.js
@@ -169,7 +169,7 @@ before(async function() {
await exec("node ../../packages/master config set master.http_port 8880");
await exec("node ../../packages/master config set master.https_port 4443");
await exec("node ../../packages/master config set master.heartbeat_interval 0.25");
- await exec("node ../../packages/master config set master.connector_shutdown_timeout 2");
+ await exec("node ../../packages/master config set master.session_timeout 2");
await exec("node ../../packages/master config set master.tls_certificate ../../test/file/tls/cert.pem");
await exec("node ../../packages/master config set master.tls_private_key ../../test/file/tls/key.pem");
@@ -189,7 +189,7 @@ before(async function() {
await exec(`node ../../packages/slave config set slave.factorio_directory ${path.join("..", "..", "factorio")}`);
await exec("node ../../packages/slave config set slave.tls_ca ../../test/file/tls/cert.pem");
- slaveProcess = await spawn("slave:", "node ../../packages/slave run", /SOCKET \| registering slave/);
+ slaveProcess = await spawn("slave:", "node ../../packages/slave run", /Started slave/);
let tlsCa = await fs.readFile("test/file/tls/cert.pem");
let controlConnector = new TestControlConnector(url, 2, tlsCa);
diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js
index 9fec97e1..6daa4d79 100644
--- a/test/lib/link/connector.js
+++ b/test/lib/link/connector.js
@@ -48,7 +48,7 @@ describe("lib/link/connectors", function() {
describe(".sendHandshake()", function() {
it("calls send on the socket", function() {
- testConnector._state = "handshake";
+ testConnector._state = "connecting";
testConnector._socket.sentMessages = [];
testConnector.sendHandshake("test", { test: true });
assert.deepEqual(
@@ -93,7 +93,11 @@ describe("lib/link/connectors", function() {
it("should emit connect on ready", async function() {
let result = events.once(testConnector, "connect");
testConnector._processHandshake(
- { seq: 1, type: "ready", data: { session_token: "x", heartbeat_interval: 10 }}
+ { seq: 1, type: "ready", data: {
+ session_token: "x",
+ session_timeout: 20,
+ heartbeat_interval: 10,
+ }}
);
await result;
clearInterval(testConnector._heartbeatId);
@@ -106,14 +110,14 @@ describe("lib/link/connectors", function() {
assert(testConnector._socket.events.size > 0, "No handlers were attached");
});
it("should throw on message received in invalid state", function() {
- testConnector._state = "new";
+ testConnector._state = "closed";
assert.throws(
() => testConnector._socket.events.get("message")("{}"),
- new Error("Received message in unexpected state new")
+ new Error("Received message in unexpected state closed")
);
});
it("should call _processHandshake on message in handshake state", function() {
- testConnector._state = "handshake";
+ testConnector._state = "connecting";
let called = false;
testConnector._processHandshake = () => { called = true; };
testConnector._socket.events.get("message")("{}");