From de48d807f13944881f6eb1bbceeb4bd05c7d3d84 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 30 Aug 2021 02:08:04 +0200 Subject: [PATCH] Refactor connector logic Refine the state flow used in the connector abstractions to better handle close while connecting and resuming. The new logic is to close the connection immediately in these two states instead of continuing trying to connect and then giving up some point later. For the connectors the handshake state is split into connecting and resuming, new is replaced by closed, and the closing state is turned into a boolean flag as it's possible that closing is desired while connecting, connected or resuming. The connecting state is not present in the server connector because there's nothing for the server to do in that connector state. The timeout logic is completely removed, once a connection attempt is started for the client it will continue to attempt to connect, attempt to resume dropped connections and attempt reconnects until either an authentication error occurs or the client explicitly requests to close the connection. A new session timeout is added to control how long to attempt to resume the connection until it's considered dead and invalidated. This allows stale connections to give up earlier, and swallow up less data while in the dropped but not yet invalidated state. For the client side sending messages while it's connecting has been changed to throw an error instead of queuing up the message and (possibly) sending it after the connection is established. Code shouldn't try to send messages without knowing if the connection it's trying to send them over is active or not. --- docs/assets/WsClientConnector.svg | 3 + docs/assets/WsServerConnector.svg | 3 + docs/assets/connection_flow.dot | 20 - docs/assets/connection_flow.drawio | 321 ++++++ docs/assets/connection_flow.svg | 102 +- docs/assets/connector_flow.drawio | 1171 ++++++++++++++++++++++ docs/assets/connector_states.svg | 3 + 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/devs/protocol.md | 268 +++++ docs/socket.md | 225 ----- packages/ctl/ctl.js | 12 +- packages/lib/config/definitions.js | 22 +- packages/lib/link/connectors.js | 264 +++-- packages/lib/schema.js | 6 +- packages/master/src/WsServer.js | 14 +- packages/master/src/WsServerConnector.js | 104 +- packages/slave/slave.js | 8 +- packages/web_ui/src/bootstrap.jsx | 3 +- packages/web_ui/src/util/websocket.jsx | 6 +- test/integration/clusterio.js | 12 +- test/integration/index.js | 4 +- test/lib/link/connector.js | 14 +- 26 files changed, 2042 insertions(+), 821 deletions(-) create mode 100644 docs/assets/WsClientConnector.svg create mode 100644 docs/assets/WsServerConnector.svg delete mode 100644 docs/assets/connection_flow.dot create mode 100644 docs/assets/connection_flow.drawio create mode 100644 docs/assets/connector_flow.drawio create mode 100644 docs/assets/connector_states.svg delete mode 100755 docs/assets/generate.sh delete mode 100644 docs/assets/reconnection_flow.dot delete mode 100644 docs/assets/reconnection_flow.svg delete mode 100644 docs/assets/server_flow.dot delete mode 100644 docs/assets/server_flow.svg create mode 100644 docs/devs/protocol.md delete mode 100644 docs/socket.md 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 @@ + + +
connected
connected
yes
yes
resuming
resuming
connecting
connecting
connect()
connect()
closed
closed

WsClientConnector

WsClientConnector
New instance
New instance
Create socket and
attach handlers
Create socket and...
Send register
Send register
Start heartbeat
Start heartbeat
close()
close()
Wait backoff delay
Wait backoff delay
socket closed
socket closed
receive hello
receive hello
receive ready
receive ready
Wait for event
Wait for event
heartbeat timeout
heartbeat timeout
socket closed
socket closed
Wait for event
Wait for event
Create socket and
attach handlers
Create socket and...
Send resume
Send resume
Wait backoff delay
Wait backoff delay
receive
hello
receive...
receive continue
receive continue
socket
closed
socket...
receive invalidate
receive invalidate
Wait for event
Wait for event
close socket
close socket
Stop heartbeat
Stop heartbeat
Sate state connected
Emit resume
Sate state connected...
Set state connected
Emit connect
Set state connected...
no
no
yes
yes
Backoff delay greater
than timeout?
Backoff delay greater...
close()
close()
Set closing flag
Set closing flag
yes
yes
no
no
Closing
flag set?
Closing...
connecting / resuming
connecting / resuming
connected
connected
Check state
Check state
Close socket
Close socket
Done
Done
no
no
Closing flag set?
Closing flag set?
Set state resuming
Emit drop
Set state resuming...
Set state connecting
Emit invalidate
Set state connecting...
yes
yes
no
no
Socket open?
Socket open?
Clear timeout and continue from Reset state
Clear timeout and continue...
no
no
yes
yes
Closing flag set?
Closing flag set?
Legend
Legend
Action
Action
Decision
Decision
Non-transient state
Non-transient state
Reset state
Emit close
Reset state...
Set state connecting
Set state connecting
Start/end state
Start/end state
close()
close()
setClosing()
setClosing()
Set closing flag
Set closing flag
connecting / resuming
connecting / resuming
connected
connected
Check state
Check state
call close()
call close()
Done
Done
Set state connecting
Emit invalidate
Set state connecting...Viewer does not support full SVG 1.1 \ 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 @@ + + +
connected
connected
yes
yes
resuming
resuming
ready()
ready()
closed
closed

WsServerConnector

WsServerConnector
New instance
New instance
Start heartbeat
Start heartbeat
heartbeat
timeout
heartbeat...
socket closed
socket closed
Wait for event
Wait for event
continue()
continue()
timeout / close()
timeout / close()
Wait for event
Wait for event
close socket
close socket
Stop heartbeat
Stop heartbeat
Send continue
Sate state connected
Attach handlers
Emit resume
Send continue...
Send ready
Set state connected
Attach handlers
Emit connect
Send ready...
close()
close()
connected
connected
resuming
resuming
Check state
Check state
Set closing flag
Close socket
Set closing flag...
Done
Done
no
no
Closing flag set?
Closing flag set?
Set state resuming
Emit drop
Set state resuming...
Legend
Legend
Action
Action
Decision
Decision
Non-transient state
Non-transient state
Reset all state
Emit close
Reset all state...
Start/end state
Start/end state
setClosing()
setClosing()
connected
connected
resuming
resuming
Check state
Check state
Set closing flag
Set closing flag
Done
Done
Continue from
reset all state
Continue from...
Continue from
reset all state
Continue from...Viewer does not support full SVG 1.1 \ 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 @@ - - - - - - -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 - - - + + +
Client
Client
receive
resume
receive...
receive
register
receive...
bad handshake
timeout
bad handshake...
Wait for handshake
Wait for handshake
Server
Server

Connection Flow

Connection Flow
WebSocket Open
WebSocket Open
bad session
bad session
valid
valid
Validate resume
Validate resume
valid
valid
bad token
bad token
Validate register
Validate register
Send hello
Send hello
Send invalidate
Send invalidate
Close connection
Close connection
Connected
Connected
Send ready
Send ready
Send continue
Send continue
previous
session
previous...
no
session
no...
bad handshake
timeout
bad handshake...
Wait for hello
Wait for hello
WebSocket Open
WebSocket Open
Send resume
Send resume
Send register
Send register
Close connection
Close connection
Connected
Connected
receive
invalidate
receive...
bad handshake
timeout
bad handshake...
receive
continue
receive...
Wait for continue
Wait for continue
Invalidate session
Invalidate session
bad handshake
timeout
bad handshake...
receive
ready
receive...
Wait for ready
Wait for readyViewer does not support full SVG 1.1 \ 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 @@ + + +
connected
connected
resuming
resuming
closed
closed
connected
connected
resuming
resuming
WsServerConnector
WsServerConnector
ready()
ready()
timeout
invalidate
timeout...
continue
continue
drop
drop
timeout
close()
timeout...
close()
close()
close()
close()
connecting
connecting
closed
closed
close()
close()
connect()
connect()
close()
close()
drop
drop
continue()
continue()
WsClientConnector
WsClientConnector

Connector States

Connector StatesViewer does not support full SVG 1.1 \ 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 @@ - - - - - - -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 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 @@ - - - - - - -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/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`. + +![Connection flow](../assets/connection_flow.svg) + + +## 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` + +![Conector States](../assets/connector_states.svg) + + +### 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. + +![WsClientConnector](../assets/WsClientConnector.svg) + + +### 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. + +![WsServerConnector](../assets/WsServerConnector.svg) + + +## 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. - -![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". -- 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")("{}");